如何忽略Celery Chord中的错误并让工作流程继续执行

我有一个包含多个task、chain和chord的Celery工作流，部分任务串行、部分并行。当前问题是chord中某个任务报错时，整个chain流程会中断，需要让流程即使遇到错误也能继续执行。

复现示例代码 #

from celery import chain, chord

from celery_app import celery_app


@celery_app.task(name="TASK_1")
def task_1():
    print("task_1")


@celery_app.task(name="TASK_2")
def task_2():
    print("task_2")


@celery_app.task(name="PARALLEL_TASK")
def parallel_task():
    # HEAVY HTTP Call
    print("parallel task")


@celery_app.task(name="ERROR_PARALLEL_TASK")
def error_parallel_task():
    print("task error")
    raise ValueError()


@celery_app.task(name="TASK_3")
def task_3():
    print("task_3")


@celery_app.task(name="TASK_4")
def task_4():
    print("task_4")


chain(
    task_1.si(),
    task_2.si(),
    chord(
        [
            parallel_task.si(),
            parallel_task.si(),
            error_parallel_task.si(),
            error_parallel_task.si(),
            parallel_task.si(),
            parallel_task.si()
        ],
        task_3.si(),
    ),
    task_4.si(),
).delay()

运行后抛出错误：
celery.exceptions.ChordError: Dependency 10719ad2-cd61-4596-bb9b-92d42d1ea967 raised ValueError()

已尝试的无效/不合适方法 #

• 配置项尝试：ignore_result=True、throws=(ValueError,)、acks_on_failure_or_timeout=True，均无法阻止流程中断
• 使用.on_error()：可行但会重复调用后续任务（如task_3被多次触发），不符合需求
• 任务内捕获异常：虽能运行，但任务实际失败却显示成功，且无法正常使用Celery重试机制

解决方案 #

方案1：配置+任务错误回调，让Chord忽略失败 #

Celery有个默认配置CHORD_PROPAGATES，默认值为True，会在子任务失败时抛出ChordError中断流程。将其设为False后，Chord的回调会在所有子任务完成（无论成功失败）后执行，同时给失败任务绑定错误日志回调，保留错误记录能力。

步骤：

1. 修改Celery配置：

# celery_app.py中添加配置
celery_app.conf.update(
    CHORD_PROPAGATES=False,  # 禁止Chord传播子任务失败的异常
)

2. 给可能失败的任务绑定错误处理任务：

@celery_app.task(name="LOG_TASK_FAILURE")
def log_task_failure(task_id, exc, args, kwargs):
    # 记录失败任务的日志，不中断流程
    print(f"任务 {task_id} 执行失败: {exc}")

# 修改后的Chain调用
chain(
    task_1.si(),
    task_2.si(),
    chord(
        [
            parallel_task.si(),
            parallel_task.si(),
            # 给失败任务绑定错误日志回调
            error_parallel_task.si().link_error(log_task_failure.s()),
            error_parallel_task.si().link_error(log_task_failure.s()),
            parallel_task.si(),
            parallel_task.si()
        ],
        task_3.si(),
    ),
    task_4.si(),
).delay()

方案2：自定义Chord回调包装器，主动过滤失败结果 #

如果需要对失败任务的结果做额外处理（比如统计失败数量），可以写一个中间包装任务，接收Chord子任务的结果列表，过滤失败项后再调用目标任务。

示例代码：

@celery_app.task(name="CHORD_CALLBACK_WRAPPER")
def chord_callback_wrapper(results):
    # 过滤失败的任务结果（Exception类型为失败结果）
    successful_tasks = [res for res in results if not isinstance(res, Exception)]
    # 可选：记录失败任务数量
    print(f"并行任务执行完成，成功{len(successful_tasks)}个，失败{len(results)-len(successful_tasks)}个")
    # 调用原回调任务task_3
    return task_3.si().delay()

# 修改后的Chain调用
chain(
    task_1.si(),
    task_2.si(),
    chord(
        [
            parallel_task.si(),
            parallel_task.si(),
            error_parallel_task.si(),
            error_parallel_task.si(),
            parallel_task.si(),
            parallel_task.si()
        ],
        chord_callback_wrapper.s(),  # 使用包装后的回调，注意用s()接收结果列表
    ),
    task_4.si(),
).delay()

方案3：用Group+Chain替代Chord（更灵活） #

如果不需要严格等待所有并行任务完成再执行后续任务，可以直接用Group替代Chord，因为Group本身不会因为子任务失败而中断整个Chain流程，只要所有子任务都被调度处理（包括失败），就会继续执行后续任务。

示例代码：

from celery import group

chain(
    task_1.si(),
    task_2.si(),
    group(
        parallel_task.si(),
        parallel_task.si(),
        error_parallel_task.si().link_error(log_task_failure.s()),
        error_parallel_task.si().link_error(log_task_failure.s()),
        parallel_task.si(),
        parallel_task.si()
    ),
    task_3.si(),
    task_4.si(),
).delay()

关于Celery是否适合该需求 #

Celery完全适合实现这种容错式工作流，默认的Chord行为是严格的（只要有一个依赖失败就中断），但通过配置调整或自定义逻辑，完全可以实现“忽略错误继续执行”的需求。具体方案可根据场景选择：

• 若需要严格等待所有并行任务完成再执行回调：选方案1或方案2
• 若不需要严格等待所有任务完成：选方案3

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