如何在Celery任务依赖流中避免任务b被重复执行？

问题分析 #

你的当前实现中，因为在两个子group里分别调用了b.s()，导致任务b被调度执行两次。要实现任务b仅执行一次，同时满足d接收a和b的输出、e接收b和c的输出，需要先让a、b、c并行执行，再将它们的结果按需分发给d和e。

注意：原任务定义中使用了results={}这种默认可变参数，这在Python中存在陷阱——默认字典会被所有任务调用共享，可能导致意外的状态污染，建议避免使用这类参数，改为显式传递参数。

解决方案代码 #

调整任务定义（推荐，避免参数陷阱） #

先修改任务参数，让依赖关系更清晰：

@app.task(bind=True)
def a(task):
    return {"key": "A"}

@app.task(bind=True)
def b(task):
    return {"key": "B"}

@app.task(bind=True)
def c(task):
    return {"key": "C"}

@app.task(bind=True)
def d(task, a_result, b_result):
    return {"key": "D", "from_a": a_result, "from_b": b_result}

@app.task(bind=True)
def e(task, b_result, c_result):
    return {"key": "E", "from_b": b_result, "from_c": c_result}

构建无重复执行的任务流 #

通过chain结合group，先并行执行a、b、c，再将结果列表中的对应值传递给d和e：

from celery import group, chain

# 第一步：并行执行a、b、c，得到结果列表 [a_res, b_res, c_res]
abc_group = group(a.s(), b.s(), c.s())

# 第二步：定义d和e的任务，从结果列表中提取各自需要的部分
post_tasks = group(
    # d接收a和b的结果（结果列表的第0、1位）
    d.s().set(args=(lambda res: (res[0], res[1]))),
    # e接收b和c的结果（结果列表的第1、2位）
    e.s().set(args=(lambda res: (res[1], res[2])))
)

# 组合成完整任务流：先执行abc_group，再执行post_tasks
celery_graph = chain(abc_group, post_tasks)

# 启动任务
res = celery_graph.delay()

兼容原任务定义的写法 #

如果不想修改原任务参数，可直接传递结果字典：

from celery import group, chain

abc_group = group(a.s(), b.s(), c.s())

post_tasks = group(
    d.s(results={"a": lambda res: res[0], "b": lambda res: res[1]}),
    e.s(results={"b": lambda res: res[1], "c": lambda res: res[2]})
)

celery_graph = chain(abc_group, post_tasks)
res = celery_graph.delay()

原理说明 #

1. abc_group会并行调度a、b、c三个任务，任务b仅被执行一次，最终返回包含三个任务结果的列表。
2. chain确保abc_group执行完成后，再调度post_tasks中的d和e任务。
3. 通过lambda res: ...从结果列表中提取d、e需要的依赖结果，实现结果复用。

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