Celery 在任务内异步运行子任务并等待结果

Question

我有一个 celery 任务，可以启动其他三个 celery 任务。我希望这些任务异步执行并等待它们完成，然后再恢复父任务。然而，子任务正在同步运行，我不知道为什么。 时，问题就开始了，

当我将 celery 从 4.4.7 升级到 5.0.0 app_celery.py

@app.task(name="app_celery.scraping_process", soft_time_limit=900, time_limit=960, max_retries=3)
def scraping_process():
    sources = ["a", "b", "c"]
    job = group((company_representation.s(src) for src in sources))
    result = job.apply_async(queue="spiders", routing_key="spiders")
    while not result.ready():
        time.sleep(5)
    
@app.task(name="app_celery.company_representation", max_retries=3)
def company_representation(source: str):
    # do something
    time.sleep(60)
    

我正在像这样运行 celery：

celery -A app_celery worker -c 8 -Q spiders -n spiders@%%h
celery -A app_celery worker -c 2 -Q companies -n companies@%%h --without-mingle --without-heartbeat -Ofair

celery==5.0.0

Answer 1

您可以将任务 ID 添加到列表中，然后执行以下操作：

def poll_job_status(active_jobs):
    if len(active_jobs) == 1:
        task = task.AsyncResult(active_jobs[0])
        if not task.ready():
            return active_jobs
    _new_active_jobs = []
    for taskid in active_jobs:
        task = task.AsyncResult(taskid)
        if task.state == "PENDING" or task.state == "RETRY":
            _new_active_jobs.append(taskid)
    active_jobs = _new_active_jobs
    return active_jobs

因此，您将迭代任务 ID 列表并检查任务是否完成。然后，如果列表为空，您就知道所有任务都已运行，您可以继续执行其他操作。示例用法是：

active_tasks_list = []
active_tasks_list.append(task.delay(args).id)
while len(active_tasks_list) > 0:
   poll_job_status(active_tasks_list)
# carry on other processes

如果您有大量要跟踪的任务，这是理想的选择。

Answer 2

您可以尝试使用 Celery 组并行调用多个任务并等待它们的结果。

@app.task(name="app_celery.scraping_process", soft_time_limit=900, time_limit=960, max_retries=3)
def scraping_process():
    sources = ["a", "b", "c"]
    tasks =[]
    for src in sources:
        tasks.append(company_representation.s(src))

    # create a group with all the tasks
    job = group(tasks)
    result = job.apply_async(queue="spiders", routing_key="spiders")
    ret_val = result.get(disable_sync_subtasks=False)
    return ret_val
    
@app.task(name="app_celery.company_representation", max_retries=3)
def company_representation(source: str):
    # do something
    time.sleep(60)

参考：http://ask.github.io/celery/userguide/groups。 html#groups