Celery工作流错误处理：Combine任务如何感知子任务错误并终止执行？

嘿，这个需求在Celery里完全能搞定，我给你分享几个实战中好用的方案，帮你实现「子任务出错就终止，无错就继续执行格式化任务」的逻辑：

方案1：用chord+link_error统一收集错误 #

如果你的Task1、Task2是并行执行的，chord是最适合的工具——它能先执行一组任务，再把所有结果（包括错误）传给后续的Combine Task。

实现步骤： #

1. 先定义一个错误收集任务，用来捕获每个子任务的错误信息；
2. 给每个子任务绑定link_error，指向错误收集任务；
3. 用group把子任务打包，再通过chord把任务组的结果和错误一起传给Combine Task。

代码示例： #

from celery import Celery, chord, group

app = Celery('task_flow', broker='redis://localhost:6379/0')

# 1. 错误收集任务：捕获子任务的错误并返回结构化信息
@app.task
def collect_task_error(error_result):
    return {
        'task_id': error_result.id,
        'error_msg': str(error_result.info),
        'task_state': error_result.state
    }

# 2. 你的业务子任务
@app.task(bind=True)
def task1(self, data):
    # 模拟错误场景
    if data < 0:
        raise ValueError("Task1：输入数据不能为负数")
    return data * 2

@app.task(bind=True)
def task2(self, data):
    # 模拟错误场景
    if data > 100:
        raise TypeError("Task2：输入数据超出范围")
    return data + 3

# 3. Combine Task：检查错误并决定后续流程
@app.task
def combine_task(task_results, collected_errors=None):
    # 先判断是否有错误
    if collected_errors and len(collected_errors) > 0:
        print(f"检测到子任务错误，终止执行：{collected_errors}")
        return {"status": "aborted", "errors": collected_errors}
    
    # 无错误则执行格式化任务
    format_results_task.delay(task_results)
    return {"status": "success", "results": task_results}

# 4. 格式化结果任务
@app.task
def format_results_task(raw_results):
    return [f"格式化结果：{item}" for item in raw_results]

调用方式： #

# 构建带错误回调的任务组
task_group = group(
    task1.s(10).link_error(collect_task_error.s()),
    task2.s(20).link_error(collect_task_error.s())
)

# 用chord执行：先跑任务组，再把结果+错误传给combine_task
chord(task_group)(combine_task.s())

方案2：自定义任务基类，全局处理错误 #

如果你的任务是串行链式执行（Task1→Task2→Combine Task），可以自定义一个任务基类，在on_failure钩子中主动把错误传递给Combine Task。

实现步骤： #

1. 自定义继承Celery.Task的基类，重写on_failure方法；
2. 子任务继承这个基类，出错时自动触发错误传递；
3. Combine Task通过Redis等存储介质收集错误（因为Celery是分布式的，内存变量无法共享）。

代码示例： #

from celery import Celery, Task
import redis

app = Celery('task_flow', broker='redis://localhost:6379/0')
# 用Redis存储错误信息，保证分布式环境下能共享
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# 1. 自定义错误跟踪任务基类
class ErrorTrackingBaseTask(Task):
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        # 从kwargs中拿到Combine Task的ID，把错误存入Redis
        combine_task_id = kwargs.get('combine_task_id')
        if combine_task_id:
            error_key = f"combine_errors:{combine_task_id}"
            redis_client.rpush(error_key, f"任务[{task_id}]出错：{str(exc)}")
        # 调用父类的on_failure，保留原有错误处理逻辑
        super().on_failure(exc, task_id, args, kwargs, einfo)

# 2. 业务子任务继承基类
@app.task(base=ErrorTrackingBaseTask)
def task1(data, combine_task_id):
    if data == 0:
        raise ZeroDivisionError("Task1：输入不能为0")
    return data * 3

@app.task(base=ErrorTrackingBaseTask)
def task2(data, combine_task_id):
    if isinstance(data, str):
        raise TypeError("Task2：输入必须是数字")
    return data + 5

# 3. Combine Task：检查Redis中的错误
@app.task(bind=True)
def combine_task(self):
    error_key = f"combine_errors:{self.request.id}"
    collected_errors = redis_client.lrange(error_key, 0, -1)
    
    if collected_errors:
        print(f"发现错误，终止流程：{collected_errors}")
        # 清理Redis中的错误键
        redis_client.delete(error_key)
        return {"status": "aborted", "errors": collected_errors}
    
    # 无错误则执行格式化任务（这里需要自己传入子任务结果，或者从Redis/数据库读取）
    format_results_task.delay([30, 25])
    redis_client.delete(error_key)
    return {"status": "success"}

调用方式： #

# 先启动Combine Task，拿到它的ID
combine_result = combine_task.delay()
# 子任务传入Combine Task的ID，出错时自动上报
task1.delay(10, combine_task_id=combine_result.id)
task2.delay(20, combine_task_id=combine_result.id)

方案3：链式任务+errbacks直接跳转错误 #

如果是严格的串行链式流程（Task1执行完才到Task2，再到Combine Task），可以给每个任务加错误回调，一旦出错就直接调用Combine Task，跳过后续正常任务。

代码示例： #

from celery import Celery, chain

app = Celery('task_flow', broker='redis://localhost:6379/0')

# 错误回调：直接把错误传给Combine Task
def forward_error_to_combine(error_result):
    combine_task.delay(errors=[f"错误来自任务[{error_result.id}]：{str(error_result.info)}"])

# 业务子任务
@app.task
def task1(data):
    if data < 0:
        raise ValueError("Task1：数据非法")
    return data * 2

@app.task
def task2(data):
    if data > 50:
        raise OverflowError("Task2：数据溢出")
    return data + 10

# Combine Task
@app.task
def combine_task(results=None, errors=None):
    if errors:
        print(f"终止执行，错误信息：{errors}")
        return {"status": "failed", "errors": errors}
    format_results_task.delay(results)
    return {"status": "success", "results": results}

@app.task
def format_results_task(results):
    return [f"最终结果：{item}" for item in results]

调用方式： #

# 构建链式任务，每个任务的link_error指向错误回调
task_chain = chain(
    task1.s(-5).link_error(forward_error_to_combine),
    task2.s().link_error(forward_error_to_combine),
    combine_task.s()
)

task_chain.delay()

方案选择建议 #

• 如果是并行子任务：优先选方案1的chord，它天然支持任务组的结果和错误聚合，代码最简洁；
• 如果是串行链式任务：方案3的chain+errbacks更直接，或者方案2的自定义基类适合需要全局统一错误处理的场景；
• 分布式环境下：尽量用Redis/数据库存储错误信息，避免内存变量无法共享的问题。

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