如何在Python中使用Redis分布式锁防止定时任务重复执行？-Python教程-PHP中文网

直接用SETNX不够安全，因其无法自动过期、缺乏唯一标识校验且加锁与设过期非原子操作，易导致死锁或误删他人锁。

如何在python中使用redis分布式锁防止定时任务重复执行？

为什么直接用 `SETNX` 不够安全？

很多同学第一反应是用 redis.setnx(key, value) 加锁，但实际生产中会出问题：如果任务执行超时，锁没释放，后续所有定时任务都会被阻塞或失败；更糟的是，如果进程崩溃、机器宕机，锁永远残留。所以必须带自动过期 + 唯一标识（防止误删别人锁）+ 原子性释放。

用 `SET key value EX seconds NX` 实现可靠加锁

Redis 2.6.12+ 支持 SET 的扩展参数，能一步完成「设值 + 过期 + 仅当不存在时设置」，比 SETNX + EXPIRE 组合更原子。关键点：

value 必须是全局唯一字符串（比如 uuid.uuid4().hex），后续删锁时靠它校验所有权
EX 时间要明显大于任务最长执行时间（例如任务通常 30s，设为 120s），避免锁提前失效引发并发
返回 True 表示抢锁成功，False 表示锁已被占，应直接退出本次任务

示例：

import redis
import uuid

r = redis.Redis()
lock_key = "task:send_daily_report"
lock_value = uuid.uuid4().hex
acquired = r.set(lock_key, lock_value, ex=120, nx=True)
if not acquired:
    print("锁已被占用，跳过执行")
    exit(0)

如何安全释放锁？不能直接 `DEL`

直接 r.delete(lock_key) 会误删其他实例的锁——比如 A 拿到锁后卡住超时，B 又抢到新锁，这时 A 恢复并删掉 B 的锁，就出问题了。必须用 Lua 脚本保证「判断 value 相等再删」的原子性：

Python Hardener

通过完整的安全审查、代码整洁重构、健壮的错误处理、正确的日志记录以及完整的文档字符串覆盖，强化 Python 脚本。

下载

unlock_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end
"""
unlock = r.register_script(unlock_script)
unlock(keys=[lock_key], args=[lock_value])

注意：unlock() 返回 1 表示删除成功，0 表示锁已不属于当前实例（比如已被续期或过期），此时无需重试。