芹菜和弦在所有任务完成之前执行

Question

我有很多任务可以同时执行，但一旦一切就绪，我想执行最后一个任务。我正在使用以下代码：

chunk_tasks = []
for index, chunk in enumerate(chunks):
    chunk_tasks.append(import_chunk.s(meta.pk))
g = group(chunk_tasks)
chord(g)(import_completed.s(meta.pk, max_lines=max_lines))

然而，看起来import_completed是在所有任务完成之前执行的。此外，import_chunk任务看起来像：

@task(bind=True, ignore_result=IGNORE_RESULTS)
def import_chunk(self, meta_pk):
    try:
        # do some stuff
    except Exception, e:
        if self.max_retries == self.request.retries:
            logger.exception('Unexpected error in import_chunk')
        raise self.retry(countdown=60, max_retries=3)

所以问题是我做错了什么？

Answer 1

Chord是一个只有在组中的所有任务都完成执行后才执行的任务。因此，它需要在其标头中的任务状态进行同步。

但是，当您将ignore_result设置为task时，工作者将不会存储任务状态并返回此任务的值。

这将导致根据您的工作流程重试任务或引发异常或任何故障。

所以，chord(add.s(i, i) for i in range(10))(tsum.s()).get()是完全有效的，并给出了情况1的结果，但它给情况2带来了一些麻烦。

情况1:

@app.task
def add(x, y):
    return x + y
@app.task
def tsum(numbers):
    return sum(numbers)

情况2:

@app.task(ignore_result=True)
def add(x, y):
    return x + y
@app.task(ignore_result=True)
def tsum(numbers):
    return sum(numbers)

因此，您必须更改ignore_result或更改任务的工作流。

来自文档：

你应该尽量避免使用和弦。尽管如此，chord是工具箱中一个强大的基元，因为同步是许多并行算法所必需的步骤