Django怎么防止Celery任务重复执行：Python结合Redis实现分布式锁
你遇到过吗？明明只发了一次任务，后台却执行了两次。这不是代码写错了，而是分布式环境下一个经典的老朋友：多个worker同时抢到了同一个活儿。
为什么Celery任务会重复执行
问题的根源在于竞争。想象一下，多个Celery worker同时从Redis队列里看到了同一个task_id，而系统又没有设置“此任务已有人处理”的标识，结果就是大家一拥而上，都开始执行同一段逻辑。这种情况在网络抖动、任务重试（retries）或者worker重启时尤其高发。
这里有个常见的误解：以为Celery的task_id自带互斥属性。其实不然，它只保证全局唯一性，并不管执行时的并发控制。Redis本身也不提供“正在执行任务”的全局视图，所以得靠我们自己来加把“锁”。
用Redis SETNX实现最简分布式锁
实现分布式锁并不复杂，无需引入重型框架，用Redis的原生命令就能搞定。核心在于这条命令：SET key value EX seconds NX。
- NX是关键：它保证了只有第一个设置值的客户端能成功，后续请求都会失败，这就实现了互斥。
- EX是保险：给锁设置一个自动过期时间。这是为了防止某个worker拿到锁后崩溃，导致锁永远无法释放，形成“死锁”。
- value是身份牌：必须使用一个唯一标识（比如
uuid.uuid4().hex）。这样在释放锁时，才能验证这是不是自己当初设置的锁，避免误删了别人的锁。
来看一个直接放在任务开头的示例：
立即学习“Python免费学习笔记（深入）”；
import redis
import uuid
r = redis.Redis()
def my_task(user_id):
lock_key = f"lock:my_task:{user_id}"
lock_value = uuid.uuid4().hex
# 尝试获取锁，有效期10秒
if r.set(lock_key, lock_value, ex=10, nx=True):
try:
# 执行业务逻辑
process_user(user_id)
finally:
# 安全释放锁：先核对value，再删除
if r.get(lock_key) == lock_value.encode():
r.delete(lock_key)
else:
# 锁已被占用，直接退出或记录日志
return
Celery task装饰器封装锁逻辑
每次在任务里手动写锁代码太繁琐，也容易遗漏。更好的做法是封装成一个可复用的装饰器。这里有两个要点：一是锁的key需要能根据任务参数动态生成；二是要考虑长任务场景，可能需要锁自动续期。
- 使用
functools.wraps来保留原函数的签名，否则Celery的任务检查工具可能无法正确识别参数。 - 锁的超时时间应略大于任务的平均执行时间。例如，任务通常耗时3秒，那么锁可以设为5秒。
- 对于执行时间不确定的长时间任务，需要考虑更复杂的锁管理策略。
一个轻量级的封装示例如下：
from functools import wraps
def distributed_lock(lock_key_func, timeout=5):
def decorator(task_func):
@wraps(task_func)
def wrapper(*args, **kwargs):
key = lock_key_func(*args, **kwargs)
lock_value = str(uuid.uuid4())
if r.set(key, lock_value, ex=timeout, nx=True):
try:
return task_func(*args, **kwargs)
finally:
if r.get(key) == lock_value.encode():
r.delete(key)
else:
return None # 或者抛出特定异常
return wrapper
return decorator
# 使用方式
@distributed_lock(lambda user_id: f"lock:send_email:{user_id}", timeout=8)
def send_email_task(user_id):
# 发送邮件逻辑
...
比锁更稳妥的方案：幂等+唯一键
锁解决的是“防止并发执行”的问题，但我们的终极目标是“防止产生重复的副作用”。很多时候，与其在并发控制上死磕，不如让任务本身具备“幂等性”——即无论执行多少次，结果都和执行一次一样。
- 数据库层面：利用唯一约束。例如，为数据表添加
unique_together联合唯一索引，或在插入时使用ON CONFLICT DO NOTHING（PostgreSQL语法）。 - 状态检查：任务执行前，先查询一次记录。比如，发消息前先查
TaskResult.objects.filter(task_id=..., status='SUCCESS')，看是否已经成功过。 - Celery配置：启用
acks_late=True和reject_on_worker_lost=True（Celery 5.3+），这可以在worker意外丢失任务时，让任务重回队列而不是被确认，减少重复入队的概率。 - Redis状态标记：在Redis中用Hash记录任务状态，例如
HSET job_status:{task_id} status "started"。任务开始时先检查这个状态。
实际上，锁和幂等设计并不是二选一的关系，它们应该组合使用：用锁控制执行入口，用幂等设计兜底最终状态，这样才最保险。
最后提一个容易踩的坑：锁的粒度。用整个task_id作为锁的key意义不大，无法防止对同一业务实体（如同一个用户、同一笔订单）的重复操作。锁的key必须落到具体的业务维度，比如user_id、order_id。另一个常见的疏忽是忘记给锁设置过期时间，一旦某个任务卡住，后续所有同key的任务都会被永久阻塞。
