贝利信息

PythonCelery任务队列实战教程_链式任务与异步执行

日期:2026-01-05 00:00 / 作者:舞姬之光
链式任务(chord)是Celery中先并发执行group内所有子任务、待全部成功后再触发callback任务的任务组合方式;其核心为chord(group, callback),依赖结果后端收集结果,callback接收按group顺序排列的结果列表。

什么是链式任务(chord)

链式任务是 Celery 中一种特殊的任务组合方式,它把一组并行执行的子任务(group)和一个在所有子任务完成后才触发的回调任务(callback)串起来。你可以把它理解成“先并发跑一批任务,等它们全干完了,再统一处理结果”。

比如:要批量下载 100 个网页,下载完后统计总字数。下载动作可以并行,但统计必须等全部下载完成才能开始——这就适合用 chord。

怎么写一个 chord 链

核心写法很简单,用 chord(group, callback)

示例代码:

from celery import Celery

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

@app.task def add(x, y): return x + y

@app.task def summarize(results): return sum(results)

构建 chord:并发执行三个 add,完成后调用 summarize

result = chord( [add.s(1, 2), add.s(3, 4), add.s(5, 6)], summarize.s() ).apply_async()

print(result.get()) # 输出 21(即 3+7+11)

注意:.s() 表示创建签名(signature),它是可序列化的任务“蓝图”,不立即执行。

异步执行的关键细节

Chord 的整个流程是完全异步的,但背后有隐含依赖逻辑:

生产中建议为 callback 加上重试机制:

@app.task(bind=True, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3})
def summarize(self, results):
    return sum(results)

常见问题与避坑提醒

实际用 chord 容易踩几个坑:

需要更灵活的聚合逻辑?可以改用 group + callback 手动触发,或者用 canvas 中的 chords 结合 apply_chunks 分批处理。