Python celery - 如何等待和弦中的所有子任务答案

【问题标题】：Python celery - how to wait for all subtasks in chordPython celery - 如何等待和弦中的所有子任务
【发布时间】：2016-11-07 11:12:09
【问题描述】：

我正在对 celery 任务进行单元测试。我有链式任务也有组，因此产生了和弦。

测试应该如下所示：

运行 celery 任务（延迟）
等待任务和所有子任务
断言

我尝试了以下方法：

def wait_for_result(result):
    result.get()
    for child in result.children or list():
        if isinstance(child, GroupResult):
           # tried looping over task result in group
           # until tasks are ready, but without success 
           pass
        wait_for_result(child)

这会造成死锁，永远重试 chord_unlock。我对任务结果不感兴趣。如何等待所有子任务完成？

【问题讨论】：

标签： python celery celery-task chord

【解决方案1】：

虽然这是一个老问题，但我只是想分享一下我是如何摆脱死锁问题的，以防万一它对某人有所帮助。

就像 celery 日志所说的那样，永远不要在任务中使用get()。这确实会造成死锁。

我有一组类似的 celery 任务，其中包括一组任务链，因此使它成为一个和弦。我通过发出 HTTP 请求使用龙卷风调用这些任务。所以我所做的是这样的：

@task
def someFunction():
    ....


@task
def someTask():
    ....


@task
def celeryTask():
    groupTask = group([someFunction.s(i) for i in range(10)])

    job = (groupTask| someTask.s())

    return job

当celeryTask()被tornado调用时，链将开始执行，someTask()的UUID将保存在job中。它看起来像

异步结果：765b29a8-7873-4b28-b05c-7e19c33e950c

这个 UUID 被返回，celeryTask() 在链开始执行之前退出（理想情况下），因此为另一个进程运行留出了空间。

然后我使用龙卷风图层来检查任务的状态。龙卷风层的详细信息可以在这个stackoverflow question

中找到

【讨论】：

【解决方案2】：

你试过和弦+回调吗？

http://docs.celeryproject.org/en/latest/userguide/canvas.html#chords

>>> callback = tsum.s()
>>> header = [add.s(i, i) for i in range(100)]
>>> result = chord(header)(callback)
>>> result.get()
9900

【讨论】：