如何在芹菜中创建和弦，其任务本身就是和弦

Question

我正在尝试在芹菜中创建一个和弦，其组中的任务是和弦本身。因此，这个想法是 - 每个单个子公司首先执行，然后这些和弦的结果最终由父任务的回调使用。这是我想做的代码

- tasks.py

import celery
from celery import chord
@celery.task
def add(x, y):
    print "add called..."
    return x + y
@celery.task
def tsum(numbers):
    return sum(numbers)
@celery.task
def gr_add(x):
    print "**** gr_add called"
    c = chord(add.subtask((i,i)) for i in range(2*x, 2*x+2))
    r = c(tsum.subtask())
    return r.get(timeout=120)

- 然后我的test-chord-chord.py文件包含

from tasks import add, tsum, gr_add
from celery import chord
c = chord(gr_add.subtask((i,)) for i in range(2))
result = c(tsum.subtask())
print result.get(timeout=5)

但是，这并不是预期的。我做得正确吗？还有另一种实现上述方法吗？

Answer 1

我终于能够找出问题所在。虽然不太确定如何解决它。罪魁祸首是上面gr_add中的回调，特别是线-r = c(tsum.subtask())。这将阻止 - 因此，运行此任务的工人将阻止。现在，当我只有一个工人时 - 以上是一个问题。当我启动多个工人时（在这种情况下是三个工人 - 因为原始和弦正在创建两个和弦），每个和弦只能掌握一个任务CELERY_PREFTCH_MULTIPLIER = 1和CELERY_ACKS_LATE=True。这三名工人在队列上捡起了任务 - 第一个拿起了第一个和弦，现在被阻止了。第二个捡起第二和弦，现在被阻止。第三个拾取了第一个和弦的第一个任务 - 计算，退出 - 未经阻塞。拾取了下一个任务 - 计算/退出 - 未阻止。第一个现在没有阻止。第二和弦的第一个任务 - 计算/退出 - 未经阻止。1或3的工人之一拾取了计算的第二个子任务，计算出，退出。现在所有这三个都已大大变化，结果按预期计算。通常 - 这是一个不好的设计 - 我应该做的是"创建另一个子任务 - 可以获得和弦的结果，如果它们还不成功，只需在一段时间后重新恢复。这些重试时间间隔之间的时间将允许工人完成单个子任务，最后可以调用主呼叫和弦的回调。

编辑1：最终能够弄清楚如何处理这一点 - 阻止外部和弦所调用的任务是一个坏主意。相反，我所做的是 - 从所谓的任务中返回异步然后创建了一个民意调查任务，以进行投票，以使结果成为成功或失败。然后在此民意调查器任务中计算结果。此任务在特定持续时间后自行重试。（这几乎就像Unlock_chord任务一样 - 我自己在由代码控制的单独任务中自己完成了此操作 - 而不是让芹菜处理）