如何在芹菜中等待异步子任务完成

Question

以下是任务：

from celery import shared_task
from celery import Celery
import time

celery = Celery("example_tasks", backend="rpc://", broker="amqp://")


@shared_task
def parent_task():
    print("Starting parent task.")
    for i in range(2):
        result = child_task.delay(i)


@shared_task
def child_task(index):
    time.sleep(5)
    print(f"Starting child task {index}.")

以下是主要节目：

from celery import Celery
import time
from example_tasks import parent_task

app = Celery("tasks", backend="rpc://", broker="amqp://")

result = parent_task.delay()

在主程序结束时，等待所有任务（包括子任务）完成的最佳方式是什么

Answer 1

TL；博士

使用list(result.collect())

collect(intermediate=False, **kwargs)

Collect results as they return.

Iterator, like get() will wait for the task to complete, but will also follow AsyncResult and ResultSet returned by the task, yielding (result, value) tuples for each result in the tree.

有关更多详细信息，请参见docs

解释

第Canvas: Designing Work-flows页描述了AsyncResult.collect()方法，该方法似乎只适用于工作流，您需要一些原语来构建结果图，但实际上Celery会在创建任务时自动找到任务父级（由self.current_worker_task）

也就是说，当您创建一个子任务时，您构建了一个结果图，AsyncResult.collect()是为您准备的