如何跟踪Celery中report_builder主任务状态并获取最终S3存储路径

问题根因 #

你当前的实现中，report_builder 任务仅负责构造工作流并触发执行，触发动作完成后任务就会结束，所以状态很快标记为SUCCESS，返回值只是子工作流的AsyncResult对象，自然无法跟踪完整工作流的执行状态。

解决方案 #

推荐以下两种常用实现方式，可按需选择：

方案1：取消外层任务封装，直接触发工作流（优先推荐）

不需要把工作流包在独立的Celery任务中，直接在触发逻辑处构造并执行工作流即可，此时拿到的AsyncResult对象天然对应整个工作流的最终状态：

# 触发逻辑侧代码，无需把工作流封装为task
def trigger_report_build():
    report_workflow = chain(
        group(gather_data_from_service_a.s()....),
        merge_data.s(),
        export_to_html.s(),
        export_to_pdf.s(),
        store_pdf_on_s3.s(),
    )
    # 此处拿到的task就是整个工作流的跟踪对象
    task = report_workflow.apply_async()
    return task

该方案无需额外配置，不会产生任务死锁风险，工作流执行成功后task.result即为S3存储地址，状态也会跟随执行流程正常流转。

方案2：保留外层任务封装，同步等待工作流执行完成

如果业务要求必须通过report_builder Celery任务触发流程，可以修改任务逻辑，让其等待整个工作流执行完成后再返回结果：

@shared_task(bind=True, ignore_result=False)
def report_builder(self, *args, **kwargs):
    report_workflow = chain(
        group(gather_data_from_service_a.s()....),
        merge_data.s(),
        export_to_html.s(),
        export_to_pdf.s(),
        store_pdf_on_s3.s(),
    )
    # 触发工作流并同步等待执行结果
    final_result = report_workflow.apply_async().get()
    return final_result

注意：该方案需要提前配置Celery参数task_acks_late = True，同时确保worker并发数足够，避免出现所有worker进程都在等待子任务执行导致的死锁问题，高并发场景优先使用方案1。

内容的提问来源于stack exchange，提问作者MichelB