Python芹菜 - 如何等待所有子任务的和弦

Question

我是芹菜任务的单元测试。 我有链式任务也有组，所以会产生和弦。

测试应该是这样的：

• 运行芹菜任务（延迟）
• 等待任务和全部任务
• 断言

我试过如下：

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) 

这创建了一个de adlock，chord_unlock被永久重试。 我对任务结果不感兴趣。 如何等待所有子任务完成？

Answer 1

你试过和弦+回拨？

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 

Answer 2

尽管这是一个老问题，我只是想分享我如何摆脱僵局问题的，只是在情况下，它可以帮助别人。

就像芹菜原木说的，从来没有在任务内使用get()。这确实会造成僵局。

我有一套类似的芹菜任务，其中包括组任务链，因此使它成为一个和弦。我使用龙卷风，通过发出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()正在被龙卷风调用时，链将开始执行，&的someTask()的UUID将在job举行。它看起来像

AsyncResult：765b29a8-7873-4b28-B05C-7e19c33e950c

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

然后我使用龙卷风层来检查任务的状态。关于龙卷风层的细节可以在这里找到stackoverflow question