芹菜-如何使用和弦块?

Question

是否可以将chord()中的任务作为块运行?目前我有一个chord实现，它可以很好地处理较小的任务列表，但对于较大的任务，它最终会创建40k - 50k个任务。我想防止这种情况，因为它似乎工作效率很低，特别是在最后试图收集所有结果的时候。

我试过这个https://stackoverflow.com/a/50019022/5918981但它不适合我。也尝试在task_name.chunks().apply_async()的apply_async()中添加回调作为替代，但这给了我这个错误:"不能添加链接到组:使用和弦"。

任何帮助将不胜感激!

Answer 1

我不知道为什么回调不能直接添加到块调用中，但是这个示例代码在类似的情况下为我工作。最重要的是我将这些块转换成一个组.

def handle_name(name):
data = {}
# handle name
return data
def do_stuff_callback(results):
# results will have the following structure in this case: [[{},{}],[{},{}],....]
pass
def do_stuff():
chunk_size = 2  # dummy value
names = [......]  # a very big list
tasks = [handle_name.chunks(names, chunk_size).group()]
chord(tasks)(do_stuff_callback.s())