打破芹菜中的无限循环答案

【问题标题】：break infinite loop in celery打破芹菜中的无限循环
【发布时间】：2021-06-02 04:36:45
【问题描述】：

我使用 celery.chord(...) 创建了一组任务和一个在组中的所有任务完成后调用的方法。

我使用 amqp 结果后端（但我想切换到 memcached）。

我的工人每秒一遍又一遍地打印这条线。我不知道如何打破这个无限循环。我可以访问 rabbitMQ Web 界面，但我找不到 ID 为“32ba5fe4-...”的内容。

[2013-03-22 14:18:26,896: INFO/MainProcess] Task celery.chord_unlock[32ba5fe4-918c-480f-8a78-a310c11d0c3a] retry: Retry in 1s
[2013-03-22 14:18:26,897: INFO/MainProcess] Got task from broker: celery.chord_unlock[32ba5fe4-918c-480f-8a78-a310c11d0c3a] eta:[2013-03-22 13:18:27.895123+00:00]

这是一个测试环境。不会丢失任何数据。

我使用 Celery 3.0.16

【问题讨论】：

标签： celery

【解决方案1】：

它不应该是一个无限循环。

celery.chord_unlock 任务检查和弦子任务是否完成调用合并回调任务。如果不是，它会安排自己在一秒钟内再次检查。和弦任务完成后，您将不再在日志中看到这些消息。

已编辑：您可以撤销 chord_unlock 任务以停止循环

celery.control.revoke('32ba5fe4-918c-480f-8a78-a310c11d0c3a')

【讨论】：

【解决方案2】：

为了进行完整性检查，我在 worker 启动时通过一个信号设置了 max_retries：

from celery.signals import worker_init

@worker_init.connect
def limit_chord_unlock_tasks(worker, **kwargs):
    """
    Set max_retries for chord.unlock tasks to avoid infinitely looping
    tasks. (see celery/celery#1700 or celery/celery#2725)
    """
    task = worker.app.tasks['celery.chord_unlock']
    if task.max_retries is None:
        retries = getattr(worker.app.conf, 'CHORD_UNLOCK_MAX_RETRIES', None)
        task.max_retries = retries

然后在我的 Celery 配置中添加一个 CHORD_UNLOCK_MAX_RETRIES 变量。

【讨论】：

【解决方案3】：

我有同样的问题。为了停止循环，我安装了flower，然后从 Web 界面的 Tasks 菜单中撤销了该任务。 Revoke 按钮位于任务详细信息页面中，单击任务的 UUID 后出现。

【讨论】：