aioredis==1.x的lock()不可靠，因其加锁后释放时直接DEL且不校验持有者身份，缺乏Lua脚本保障的“验证+删除”原子性，易致误删他人锁；而redis.asyncio（aioredis 2.x）已修复此问题并支持自动续期与token校验。
为什么直接用 aioredis.Redis.lock() 不可靠
因为 aioredis（特指 aioredis==1.x）的 lock() 方法底层调用的是 Redis 的 SET key value EX seconds NX，但它**不校验锁持有者身份**，且释放时直接 DEL，存在误删风险。更关键的是：它没有实现原子性的「校验+删除」（即 Lua 脚本 unlock），在多实例竞争下极易出现锁被错误释放的问题。
如果你用的是 aioredis>=2.0（已重命名为 redis，模块为 redis.asyncio），它的 Lock 类已修复该问题，但接口和行为与 1.x 不兼容——所以必须先确认你实际用的版本：
-
pip show aioredis→ 若输出Version: 1.3.1，则仍在用旧版，不能直接信任其lock() -
import aioredis; print(aioredis.__version__)→ 同上
手写一个带 owner 校验的异步分布式锁
核心是两个原子操作：加锁用 SET key value EX seconds NX，解锁用 Lua 脚本比对 value（即唯一 owner token）再删。token 必须全局唯一（推荐用 uuid4().hex），避免不同协程误删。
下面是一个最小可用实现（适配 aioredis==1.3.1）：
import aioredis
import uuid
import asyncio
<p>class AsyncRedisLock:
def <strong>init</strong>(self, redis, key, timeout=30):
self.redis = redis
self.key = key
self.timeout = timeout
self.token = None # 懒生成，避免无谓 uuid 调用</p><pre class="brush:php;toolbar:false;">async def acquire(self, blocking=True, blocking_timeout=None):
if self.token is not None:
return False # 已持有锁，不可重入
token = uuid.uuid4().hex
# 使用 SET 命令的 NX+EX+VALUE 原子性保证
ok = await self.redis.execute('SET', self.key, token, 'EX', self.timeout, 'NX')
if ok == b'OK':
self.token = token
return True
if not blocking:
return False
# 阻塞等待：轮询 + sleep，生产环境建议加 jitter 和最大重试
while blocking:
await asyncio.sleep(0.1)
ok = await self.redis.execute('SET', self.key, token, 'EX', self.timeout, 'NX')
if ok == b'OK':
self.token = token
return True
return False
async def release(self):
if not self.token:
raise ValueError("Cannot release unacquired lock")
# Lua 脚本确保只有 own token 才能删
script = """
if redis.call("GET", KEYS[1]) == ARGV[1] then
return redis.call("DEL", KEYS[1])
else
return 0
end
"""
ret = await self.redis.eval(script, keys=[self.key], args=[self.token])
if ret != 1:
raise RuntimeError(f"Failed to release lock {self.key}: token mismatch or key missing")
self.token = None
注意点：
-
blocking_timeout参数这里没实现，如需硬超时，得在外层加asyncio.wait_for() - 轮询间隔
0.1s是平衡延迟与 Redis 压力的经验值，高并发场景可动态调整 - 务必用
self.token记录本次获取的 token，不能每次release都新生成
用 redis.asyncio（新版）更省心但要改依赖
如果你能升级到 redis>=4.2.0（即原 aioredis 2.x 的继任者），直接用内置 Lock 即可，它默认使用 Lua 解锁、支持自动续期（extend）、且线程/协程安全：
import redis.asyncio as redis
<p>r = redis.Redis(host='localhost', port=6379, db=0)
lock = r.lock("my_resource", timeout=60, blocking_timeout=5)</p><p>async def do_work():
if await lock.acquire():
try:</p><h1>执行临界区逻辑</h1><pre class="brush:php;toolbar:false;"> await asyncio.sleep(1)
finally:
await lock.release() # 自动校验 owner
关键差异：
-
timeout是锁自动过期时间（秒），不是阻塞等待时间 -
blocking_timeout是 acquire 时最多等多久，超时返回False - 内部 token 是随机字符串，unlock 通过 Lua 严格校验，无需手动管理
- 若任务执行时间可能超过
timeout，可调用await lock.extend(30)续期
最容易被忽略的三个坑
分布式锁不是加了 SET NX 就万事大吉：
- Redis 实例单点故障？锁失效 → 生产必须用 Redis Sentinel 或 Cluster，并确认 client 支持 failover 后自动重连
- 锁超时时间设太短，业务没做完锁就丢了 → 要根据 P99 业务耗时 + 安全余量设定，别拍脑袋写 5 秒
- 忘记异常路径下的
release→ 务必用try/finally或异步上下文管理器（async with lock:），新版redis.asyncio.Lock支持后者，旧版需自行封装
真正难的从来不是写几行 Lua，而是把超时、续期、故障转移、监控告警串成闭环。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
