链式任务(chord)是Celery中先并发执行group内所有子任务、待全部成功后再触发callback任务的任务组合方式;其核心为chord(group, callback),依赖结果后端收集结果,callback接收按group顺序排列的结果列表。
链式任务是 Celery 中一种特殊的任务组合方式,它把一组并行执行的子任务(group)和一个在所有子任务完成后才触发的回调任务(callback)串起来。你可以把它理解成“先并发跑一批任务,等它们全干完了,再统一处理结果”。
比如:要批量下载 100 个网页,下载完后统计总字数。下载动作可以并行,但统计必须等全部下载完成才能开始——这就适合用 chord。
核心写法很简单,用 chord(group, callback):
示例代码:
from celery import Celeryapp = 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 容易踩几个坑:
如 Redis、RabbitMQ、Database)来收集 group 结果。没配 result_backend?chord 会卡住不动需要更灵活的聚合逻辑?可以改用 group + callback 手动触发,或者用 canvas 中的 chords 结合 apply_chunks 分批处理。