Django怎么防止Celery任务重复执行_Python结合Redis实现分布式锁

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的任务都会被永久阻塞。