解决 Celery 任务重复执行：使用 Celery Once 实现分布式去重

在使用 Celery 时，有时会观察到同一个任务被执行两次。常见场景是相同的任务在不同 Worker 中几乎同时（毫秒级差异）被触发。此前常误以为是业务逻辑处理不当，但经查证，这在使用 Redis 作为 Broker 的环境中较为普遍。若不想替换现有的 Redis Broker，一种可行的解决方案是在任务执行时引入分布式锁机制。

查阅 Celery 相关 Issue 发现，社区中有人通过 Redis 自行实现分布式锁，也有人推荐使用 Celery Once。经过对比，Celery Once 的功能特性更贴合当前需求，因此决定采用该方案。

Celery Once 同样基于 Redis 加锁机制实现。它在 Celery 的 Task 类基础上扩展了 QueueOnce 类，提供任务去重功能。使用时，需将自定义任务的 base 参数设置为 QueueOnce。

@task(base=QueueOnce, once={'graceful': True})

once 参数用于定义遇到重复任务时的处理策略。默认情况下 graceful 为 False，此时 Celery 会抛出 AlreadyQueued 异常；若手动设置为 True，则静默忽略重复任务。

此外，若需手动指定任务的唯一标识 Key，可通过 keys 参数进行配置。

@celery.task(base=QueueOnce, once={'keys': ['a']})
def slow_add(a, b):
    sleep(30)
    return a + b

具体实施步骤如下：

第一步，安装依赖库。

pip install -U celery_once

第二步，在 Celery 配置中添加相关设置。

from celery import Celery
from celery_once import QueueOnce
from time import sleep

celery = Celery('tasks', broker='amqp://guest@localhost//')
celery.conf.ONCE = {
  'backend': 'celery_once.backends.Redis',
  'settings': {
    'url': 'redis://localhost:6379/0',
    'default_timeout': 60 * 60
  }
}

第三步，修改任务调用的 delay 方法。

example.delay(10)
# 修改为
result = example.apply_async(args=(10))

第四步，调整 task 装饰器的参数配置。

@celery.task(base=QueueOnce, once={'graceful': True, keys': ['a']})
def slow_add(a, b):
    sleep(30)
    return a + b

参考链接 https://github.com/cameronmaske/celery-once

至此，关于使用 Celery Once 防止 Celery 重复执行同一任务的介绍完毕。更多相关内容可搜索本站往期文章或继续浏览后续文章，感谢支持。