Django怎么防止Celery任务重复执行_Python结合Redis实现分布式锁-Python教程-PHP中文网

Celery任务重复执行是因分布式环境下多worker竞争同一任务且无互斥锁，Redis仅保证task_id唯一不保证互斥；需用SETNX实现带过期时间的分布式锁，并结合幂等设计保障最终一致性。

django怎么防止celery任务重复执行_python结合redis实现分布式锁

为什么Celery任务会重复执行

不是代码写错了，而是分布式环境下天然存在的竞争问题：多个worker同时从Redis队列里取到同一个task_id，又没加锁，结果都开始执行。尤其在retries开启、网络抖动、worker重启时特别常见。

Redis本身不提供“执行中任务”的全局视图，Celery的task_id只保证唯一性，不保证互斥——这点很多人误以为它自带防重。

用Redis SETNX实现最简分布式锁

不用引入额外库，靠Redis原生命令就能落地。核心是SET key value EX seconds NX：只有key不存在时才设值，并自动过期。

NX确保只有一个worker能抢到锁
EX避免死锁（比如worker崩溃没释放）
value必须是唯一标识（如uuid.uuid4().hex），释放锁时要校验，防止误删别人锁

示例（放在task开头）：

import redis
import uuid
<p>r = redis.Redis()</p><p>def my_task(user_id):
lock_key = f"lock:my_task:{user_id}"
lock_value = uuid.uuid4().hex</p><h1>尝试上锁，10秒超时</h1><pre class="brush:php;toolbar:false;">if r.set(lock_key, lock_value, ex=10, nx=True):
    try:
        # 执行业务逻辑
        process_user(user_id)
    finally:
        # 安全释放：先GET再DEL，且只删自己的value
        if r.get(lock_key) == lock_value.encode():
            r.delete(lock_key)
else:
    # 锁已被占，直接退出或记录日志
    return

testing-python

使用pytest编写和评估有效的Python测试。适用于编写测试、审查测试代码、调试测试失败或提高测试覆盖率。

下载

Celery task装饰器封装锁逻辑

每次手动写锁太容易漏，直接封装成可复用的decorator。注意两点：一是锁key要能动态拼接参数，二是必须支持timeout和自动续期（长任务场景）。

用functools.wraps保留原函数签名，否则Celery inspect查不到参数
锁超时时间建议略大于任务预估耗时，比如任务通常3秒，设为5秒
如果任务可能超过锁超时（如IO阻塞），需配合redis.lock的extend或用Redlock算法

轻量封装示意：

from functools import wraps
<p>def distributed_lock(lock_key_func, timeout=5):
def decorator(task_func):
@wraps(task_func)
def wrapper(*args, <strong>kwargs):
key = lock_key_func(*args, *<em>kwargs)
lock_value = str(uuid.uuid4())
if r.set(key, lock_value, ex=timeout, nx=True):
try:
return task_func(</em>args, </strong>kwargs)
finally:
if r.get(key) == lock_value.encode():
r.delete(key)
else:
return None  # 或 raise Ignore
return wrapper
return decorator</p><h1>使用</h1><p>@distributed_lock(lambda user_id: f"lock:send_email:{user_id}", timeout=8)
def send_email_task(user_id):
...
</p>