直接用 setnx 不安全，因其不支持原子性过期设置，setnx + expire 组合非原子操作易致锁残留；应使用 set 命令一次性带 nx 和 ex 参数，并配唯一 value 用于解锁校验。
为什么直接用 setnx 在 aioredis 里不安全
因为 aioredis（尤其是 v1.x）的 setnx 不支持原子性过期设置，单独调用 setnx + expire 会因网络延迟或协程切换导致锁残留——比如 setnx 成功但 expire 没发出去，锁永远不释放。
正确做法是用 Redis 的 SET 命令带 nx 和 ex 参数一次性完成，且必须配合唯一 value（如随机 UUID）用于后续校验解锁权限。
-
await redis.set("lock:key", "uuid-abc123", expire=30, nx=True)是推荐写法（v1.x 支持） - v2+ 的
aioredis.Redis实例要用set("lock:key", "uuid-abc123", ex=30, nx=True) - 别用
setnx+expire组合，这不是原子操作
如何安全地释放分布式锁
解锁不能简单 del，否则可能误删别人持有的锁。必须先比对 value 是否匹配，再删除——这个过程得原子执行，只能靠 Lua 脚本。
在 aioredis 中，用 eval 执行一段固定脚本是最稳妥方式：
lua_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
调用时传入 key 和自己的 token：
await redis.eval(lua_script, 1, "lock:key", "uuid-abc123")- 返回
1表示成功释放，0表示锁已失效或不属于当前客户端 - 不要自己拼接
get+del，中间可能被其他协程抢占
锁自动续期（heartbeat）怎么实现
如果业务逻辑执行时间不确定，超过锁过期时间会导致提前释放，引发并发冲突。需要后台任务定期刷新 TTL，但必须只刷新“自己持有”的锁。
核心是：每次续期前先 get 当前 value，确认仍是自己的 token，再用 expire 延长——注意这不是原子操作，所以续期间隔要远小于过期时间（比如设 expire=30s，每 10s 续一次）。
- 用
asyncio.create_task()启一个守护协程，在锁获取成功后启动 - 续期失败（
get返回 None 或不等于本地 token）说明锁已被抢走，应主动退出业务逻辑 - 别依赖
ttl返回值判断是否过期——它可能在你读完之后立刻过期
实际封装成可复用的异步上下文管理器
把上述逻辑收拢成一个 AsyncRedisLock 类，避免每次手写 Lua 和续期逻辑。关键点在于：构造时只存参数，__aenter__ 尝试加锁并启动 heartbeat，__aexit__ 安全释放。
示例片段：
class AsyncRedisLock:
def __init__(self, redis, key, timeout=30, retry_delay=0.1):
self.redis = redis
self.key = key
self.timeout = timeout
self.token = str(uuid4())
self._heartbeat_task = None
async def __aenter__(self):
while not await self.redis.set(self.key, self.token, ex=self.timeout, nx=True):
await asyncio.sleep(self.retry_delay)
self._start_heartbeat()
return self
使用时：
async with AsyncRedisLock(redis, "order:123") as lock:- 内部自动处理加锁、续期、释放全流程
- 注意：若业务中存在 long-running IO（如调用外部 API），需确保 heartbeat 协程不被阻塞——它得独立运行在 event loop 上
真正麻烦的不是加锁本身，而是锁生命周期和业务逻辑的耦合——token 一致性、续期时机、异常路径下的清理，漏掉任一环都可能让系统进入不可预测状态。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
