做后端时间久了你会发现，任务队列这事儿，难的不是定义任务，而是怎么编排任务。我用 Celery 的时间也快三年了，一直在和 canvas 这套 API 打交道，特别是 group 分组任务。刚开始接手一个离线报表系统时，几十个城市的统计任务全部靠手写循环发送，代码丑、排查难、结果还老对不上号。后来沉下心把 Celery canvas 里的 group、chain、chord、chunks 这几个原语捋清楚，才发现原来官方早就给了成套的编排方案，只是没人带着你把这些点串起来。

这篇内容就是围绕 Celery canvas 中的 group 任务类型展开的。我会把 group 能解决什么问题、底层怎么运作、如何上手写 Demo、生产环境里最常见的坑，以及和 chain、chord 怎么联动，全部掰开揉碎讲一遍。无论你是刚接触分布式任务队列，还是已经在用 Celery 但只会 send_task 的老伙计，这篇都能帮你把任务编排的思路理清楚。

1. 为什么需要 group：从“一堆任务怎么发”到“一批结果怎么收”

1.1 最原始的做法：循环发送任务然后各自查结果

假设你有一个批量计算任务，比如系统里有 100 个店铺，需要统计每个店铺当天的订单汇总。最直觉的写法是什么？一个 for 循环，遍历店铺 ID，然后逐个调用apply_async发送任务，最后再手动收集每个任务的AsyncResult。伪码大概是这个样子：

results = [] for shop_id in shop_id_list: async_result = calculate_shop_summary.apply_async(args=[shop_id]) results.append(async_result) # 后面某个时刻，再去逐个 results[i].get() for async_result in results: data = async_result.get(timeout=10) # 拿到数据再去整合

这个写法能用吗？能。但它有两个问题在真实项目里非常要命。

第一是代码层面散。任务的发送逻辑和结果收集逻辑被拆散到不同的函数里，一旦中间夹了其他业务逻辑，后面的人读代码根本看不出来这 100 个任务是一个逻辑批次。第二是异常处理难做。假设第 50 个任务因为店铺数据异常失败了，你要在收集阶段写一堆判断，才能知道是哪一批里的哪个任务出了问题。更麻烦的是如果这批任务是一次“全量重算”，中间失败了几个，你到底重发哪几个？靠手动维护AsyncResult列表的话，这个状态管理很快就会失控。

1.2 group 的定位：一次性发送一批任务，统一拿一个句柄

Celery canvas 把这个问题提升到了“声明式”的层面。你不再需要一个一个地apply_async，而是先构造一组任务的签名，然后丢给 group，让 group 帮你去分发。这段代码是等价的：

from celery import group job = group(calculate_shop_summary.s(shop_id) for shop_id in shop_id_list) result = job.apply_async()

注意，group(...)并不会真的发送任务。它只是构造了一个任务组的描述对象。当你调用apply_async()的时候，Celery canvas 才会把里面的任务签名逐个打包发往消息队列。

result这个对象也不再是一堆AsyncResult的散列表，而是一个 group 级别的结果句柄。你可以针对这个句柄进行统一的join()、get()、save()、forget()操作。打个生活化的比方：普通发送像是你自己跑遍全城去发传单，发完还得挨个记着哪些人拿到了；group 就是你把一摞传单交给了快递站，快递站统一派送，最后给你一张签收总表。

1.3 group 解决的核心痛点：结构化编排

我在项目中体会到，group 真正的价值不是“帮我少写几行 for 循环”，而是它把一批任务提升到了一个可操作的整体。

有了 group 之后，你可以做三件事：第一，作为一个整体做结果收集，result.get()拿到的就是按任务签名顺序排列的结果列表；第二，作为一个整体设置策略，比如group.apply_async(timeout=30)，超时控制可以在组级别统一配置；第三，作为编排单元与其他原语组合，因为group(...)对象本身就是一个可嵌套、可继续参与 chain/chord 编排的签名。

注意理解这个思维切换：过去你关注的是“单个任务”，用了 group 之后，请把注意力提升到“一批任务”这个抽象层次。很多 Celery 用得费劲的开发者，卡住的点往往就在这里。

2. group 的运作机制与核心 API 详解

2.1 从 signature 开始理解：任务与签名的区别

在 canvas 里，到处都在说 signature，也就是.s()这个东西。它到底是什么？

task.s(arg)不等于task(arg)。后者是直接调用任务函数，同步执行；前者是创建一个“签名的描述对象”，里面记录着任务名、参数、选项，但还没有真正进入执行状态。你可以把 signature 理解为一张“提货单”，上面写好了要从仓库里取什么货、送到哪里、用哪种配送方式，但货物还没出库。

group 接收的就是一组这样的签名。它只负责编排这些“提货单”，不关心具体执行细节。

# 定义一个测试任务 @app.task def add(x, y): return x + y # 两种签名写法 sig1 = add.s(1, 2) # 位置参数 sig2 = add.s(x=1, y=2) # 关键字参数 # 也可以部分绑定参数，后面再补 sig3 = add.s(1) # 先绑一个 sig3.args += (2,) # 或者合并签名时再补

这个灵活性很重要。实际业务里经常遇到“统一参数 + 各自差异参数”的组合，签名机制能让你把公共参数提前绑定好，差异参数在生成 group 时动态补齐。

2.2 group 的返回值：apply_async 之后你拿到的是什么

group.apply_async()返回的是一个GroupResult对象。别看它名字里带 Group，它本质上是对多个 AsyncResult 的封装，内部维护着一个任务 ID 列表。

这个对象有几个关键属性和方法：

通常我习惯直接调用result.get(timeout=10)。如果里面某个任务失败了，默认propagate=True，那么get()会抛出这个任务对应的异常；如果设置propagate=False，则失败任务的位置会是异常对象，其他任务结果照常返回。

2.3 结果的有序性：你以为的并发，返回时是有序的

很多人第一次用 group 时都会有个误解：既然是并发执行，那结果列表的排序是不是乱的？

不是。result.get()返回的列表顺序，永远与你创建 group 时传入的签名顺序一致。这个特性非常实用，因为它意味着你可以安全地把第 i 个结果对应到第 i 个输入，不用自己维护映射关系。

# 输入是有序的店铺列表 signatures = [calculate.s(shop_id) for shop_id in shop_id_list] # 结果列表与输入列表顺序一一对应 result = group(signatures).apply_async() res_list = result.get(timeout=30) # 直接 zip 使用 for shop_id, res in zip(shop_id_list, res_list): ...

这个有序性是收官收得非常舒服，尤其做报表拼接、批量导出这类场景，少写一大堆“手动匹配”的代码。

2.4 参数细节：timeout、propagate、interval

group 级的get()/join()有一些细节参数，平时文档里写得很简略，实战中却直接决定着你的程序是 3 秒跑完还是 3 分钟卡死。

先说timeout。这是指等待所有子任务完成的最长秒数，到点还没完成就抛TimeoutError。注意，它不代表子任务会被终止，子任务可能还在 worker 上跑。如果你在 Web 请求里使用result.get()，一定要设置合理的 timeout，否则这个请求会一直挂着，连接池迟早被耗尽。

再说interval。get()底层并不是一次性阻塞到结果出来，而是每隔一段时间去后端轮询一次。默认的interval=0.5，也就是每 0.5 秒检查一次。如果你有 500 个任务，执行时间整体约 2 秒，这个默认值完全够。但如果你用的是非默认的结果后端，轮询频率和网络开销之间就得权衡一下。

最后是propagate。在生产环境，我建议大多数场景把它设为 False，然后手动在结果列表里过滤异常。原因很简单：一个 100 任务的 group，可能只有 1 个任务因为脏数据失败了，如果直接propagate=True抛异常，剩下 99 个正常结果你也拿不到，还得额外重查，非常浪费。

res_list = result.get(timeout=30, propagate=False) for idx, res in enumerate(res_list): if isinstance(res, Exception): # 单独处理失败任务，优先重试 retry_tasks.append(origin_list[idx])

3. 实操：用 group 实现一个批量离线计算任务管道

3.1 环境准备：Celery 实例与 Redis 后端

纸上谈兵讲了不少，直接来一个能跑的完整 Demo。新建一个项目目录，里面放两个文件：tasks.py和main.py。

tasks.py负责定义 Celery 实例和任务函数，main.py是调用方，负责构造 group 并收集结果。先看任务定义：

# tasks.py from celery import Celery app = Celery( "offline_demo", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1", ) app.conf.update( task_serializer="json", result_serializer="json", accept_content=["json"], timezone="Asia/Shanghai", enable_utc=True, task_track_started=True, ) @app.task(bind=True, max_retries=3, default_retry_delay=5) def calculate_shop_summary(self, shop_id: int, factor: float = 1.0): """模拟一个耗时的店铺数据汇总任务""" # 模拟 IO 密集型耗时 import time time.sleep(2) # 模拟可能的脏数据失败 if shop_id % 17 == 0: raise ValueError(f"shop {shop_id} data inconsistent") return {"shop_id": shop_id, "total": round(shop_id * 3.14 * factor, 2)}

代码里的bind=True可以把任务实例绑定进来，重试时用self.retry()。max_retries=3这个参数配合 group 使用特别有效，可以让个别任务在 worker 内部自动重试，不影响组层面的结果收集。

3.2 构造 group 并提交任务

再看main.py：

# main.py import time from celery import group from tasks import calculate_shop_summary def run_batch(shop_ids, factor=1.0): # 构造签名列表：公共参数 factor 提前绑定，差异参数 shop_id 逐个传入 sigs = [calculate_shop_summary.s(shop_id, factor=factor) for shop_id in shop_ids] # 创建 group 对象并异步提交 group_result = group(sigs).apply_async() print(f"Group Task ID: {group_result.id}") print(f"Child Task Count: {len(group_result.children)}") return group_result if __name__ == "__main__": # 生成 1-30 的店铺 ID 列表 shop_ids = list(range(1, 31)) start = time.perf_counter() result = run_batch(shop_ids, factor=1.2) # 统一收集结果，设置 propagate=False 避免单个失败影响整体 res_list = result.get(timeout=60, propagate=False) elapsed = time.perf_counter() - start print(f"Batch completed in {elapsed:.2f}s") print(f"Success: {sum(1 for r in res_list if not isinstance(r, Exception))}/{len(res_list)}") # 打印失败的任务 failed = [(sid, res) for sid, res in zip(shop_ids, res_list) if isinstance(res, Exception)] for sid, exc in failed: print(f" Failed shop_id={sid}: {exc}")

3.3 运行与验证：并发效果和结果收集

先启动 worker：

celery -A tasks worker --loglevel=info --pool=prefork --concurrency=8

然后运行主程序：

python main.py

你会看到类似这样的输出：

Group Task ID: 9d2f8c8e-3b1a-4c5a-9b4f-8c1e6d7a2b3c Child Task Count: 30 Batch completed in 9.42s Success: 28/30 Failed shop_id=17: ValueError("shop 17 data inconsistent") Failed shop_id=34: ValueError("shop 34 data inconsistent")

这里有个很直观的验证点：30 个任务，每个任务time.sleep(2)，如果串行要 60 秒。但因为并发度是 8，整体耗时约 9.4 秒，说明 group 确实是并发执行的。这就是为什么批量任务用 group 而不是 for 循环顺序发送的底气所在。

3.4 为什么这个 Demo 的设计值得借鉴

这个 Demo 虽然简单，但几个设计习惯是经过生产检验的。

第一，公共参数提前绑定，差异参数最后补齐。factor=1.2在签名构造时统一传入，shop_id则在列表推导式里逐个传入。这样后面如果要在不同业务模块里复用，改动成本极低。

第二，propagate=False + 手动过滤失败。假设这里用默认的 propagate=True，一个店铺数据异常，整个组的 get() 直接抛异常，主流程就得中断，其他 29 个正常结果全部丢失。手动过滤后，失败的可以进入重试队列，成功的直接落库，容错性明显更好。

第三，在任务内部做重试。max_retries=3是针对偶发抖动、网络闪断这类异常的兜底，而业务脏数据异常并不适合在任务内部死磕，因为重试 3 次也一样会失败。所以任务的异常设计要分清“可重试”和“不可重试”，不可重试的尽早抛给上层处理。

4. 生产环境中踩过的 group 坑：从现象到根因的排查链路

4.1 坑一：任务偶发丢失，结果列表少了几项怎么办

某次压测时发现，100 个任务发送出去，最后get()到的结果只有 97 个，且没有超时。这个现象非常让人抓狂，因为任务队列看起来一切正常，worker 日志里也没有报错。

排查链路是这样的：先看 worker 日志，发现确实只处理了 97 个任务。接着看 Redis 里的队列长度，发送后立刻查询，发现消息入队 100 条。然后看 result 后端，发现那 3 个丢失任务对应的结果键从未被写入。再回头检查任务本身，发现其中 3 个任务抛了异常，但异常被配置项task_acks_on_failure=False且 worker 默认不把失败任务标记为“已确认”，导致消息不断重试？不是，仔细看日志发现异常直接中断了任务执行但 worker 进程没有崩溃。

最终根因是：任务函数里某段代码依赖外部接口，偶发超时抛错，而异常发生后 worker 默认行为是“确认消息并记录失败”。但那个环境里配置了task_acks_late=True，消息并没有被 ACK，同时任务也没有写失败结果到 backend，于是消息被重新投递，但同样的错误再次发生。看起来就像是“结果丢失”。

解决方式分两层：业务上给外部依赖调用加重试；框架上把失败结果写回后端，防止排查时误判为丢失。排查过程中反复对比任务日志、消息队列、结果后端三方数据，抓到了根因。

技巧：遇到 group 结果不齐，第一时间对比“worker 日志完成数”、“队列消费数”、“结果后端写入数”三个数字。三者不一致，问题大概率出现在任务执行或 ACK 配置上。

4.2 坑二：大 group 导致 broker 压力陡增，发送端的处理

有一次批量业务上线，单批任务数量直接上到 5 万。group 对象倒是构造出来了，但apply_async发送时，Redis broker 的 CPU 瞬间飙到 90%，消息积压严重，其他业务的延迟跟着遭殃。

原因很好理解：group 本身不会“一次性发送”，它会逐个把子任务发到 broker，5 万个任务一条条入队，瞬间写入压力全部压在 Redis 上。

解法不是不用 group，而是给 group 加“限流”。Celery canvas 提供了chunks，它是 group 的“分片版”，每片指定大小，逐个发送。比如每 500 个签名为一个 chunk，5 万任务就变成 100 个 chunk，每个 chunk 再内部去并发执行子任务。用法和 group 几乎一样，只是构造方式变了。

from celery import chunks # 每 500 个任务为一组 chunk_job = chunks(calculate_shop_summary.s(shop_id) for shop_id in shop_ids) | group.s(500)

这段代码的实际效果是：先生成一个个签名流，然后按 500 的分片大小切片，每个分片作为一个 group 子任务。worker 消费时拿到的是“一个分片”，然后分片内部再并发执行。broker 的压力从“瞬间 5 万条”降为“持续稳定写入”，调度系统反而更健康。

4.3 坑三：结果收集超时，单个任务跑飞了

还有一次，一个 group 里有任务被外部接口卡住，单任务跑了 20 分钟。主调用方设置的timeout=30，于是get()抛了超时。问题在于，超时任务并没有终止，它还在 worker 上继续跑，结果就是主流程已经判定失败，worker 还在空转消耗资源。

这个问题的根因在于 group 的超时控制。官方文档没有把话说透：group 的 timeout 只是等待超时，并不负责杀掉超时子任务。

正确做法是在设计任务时就要考虑“任务必须能自然结束”。所有外部调用都要设置 socket 超时和业务超时，杜绝任务永远悬挂。对于确实无法预估时长的任务，可以在组级超时后主动revoke这些子任务。

try: result = group_result.get(timeout=30) except TimeoutError: # 收集超过 30 秒仍未完成的任务 pending_results = [r for r in group_result.children if not r.ready()] for r in pending_results: r.revoke(terminate=True, signal='SIGKILL') raise

这里revoke(terminate=True)会强制杀掉对应 worker 进程里的任务执行线程，属于比较暴力的手段，能用在极少数非杀不可的场景。平时我更建议让所有任务都能在 30 秒内自然退出，这样团队的稳定性成本会低很多。

4.4 坑四：嵌套 group 时结果不对称，children 层级变深

group 是可以嵌套的，比如：

inner_1 = group(add.s(1, 2), add.s(3, 4)) inner_2 = group(add.s(5, 6), add.s(7, 8)) outer = group(inner_1, inner_2)

这种嵌套在 canvas 上是合法的，但outer.get()返回的结果并不是[3, 7, 11, 15]这样的扁平列表。实际上外层结果列表里是两个内层 group 的GroupResult对象，而不是最终计算值。

如果这是你想要的结构化结果，没问题。但如果你把嵌套当成简单的“把一堆任务归在一起”，这里就会踩坑。我用过一次嵌套 group 做多级汇总，结果收集时发现拿了一堆对象，还以为是 bug，后来才确认 canvas 就是按层级返回的。

解法有两个思路：一是扁平化构造，外层直接用所有叶子签名，不搞嵌套；二是明确用chain配合group来做层级编排，而不是单纯地嵌套 group。具体参考第 5 节。

5. 进阶：让 group 与 chain、chord 组合出真正的任务编排管线

5.1 chain：串行依赖 + group 并发分支

实际业务里，一个完整流程往往是“先做 A，然后并发做 B1/B2/B3，最后再做 C”。A 和 C 是串行依赖的，B 组则是可以并发跑的。这个结构用 chain + group 组合最自然。

from celery import chain, group def build_pipeline(shop_ids): return chain( prepare_data.s(), # 第一步：准备数据 group(calculate_shop_summary.s(sid) for sid in shop_ids), # 第二步：并发计算 aggregate_result.s(), # 第三步：汇总结果 ) pipeline = build_pipeline(shop_ids) result = pipeline.apply_async()

这段代码的含义是：先把公共数据准备好，然后并行计算每个店铺的汇总，最后把所有汇总结果代入聚合任务。中间那个 group 的执行结果是并发完成的，但整个流程依然按照“准备好 -> 算完 -> 聚合成”的顺序推进。

关键点在于：group 前一个签名（prepare_data.s()）的返回值会成为 group 内所有任务的统一输入。实际使用中要注意，group 内每个任务是否接受这个输入，以及是否需要对它做过滤。

5.2 chord：等 group 全部完成后再执行回调

chain 和 group 的组合虽然能实现“先并发再汇总”，但如果聚合任务需要 group 的返回结果作为参数，强烈建议用 chord 而不是手写 chain。

chord 的逻辑设计得很贴合业务直觉：一个 header（通常是 group）加一个 body（回调任务），header 全部执行完，所有结果汇总成列表，作为参数传给 body。

from celery import chord callback = aggregate_result.s() header = [calculate_shop_summary.s(sid) for sid in shop_ids] workflow = chord(header)(callback)

这样写的好处是回调任务拿到的一定是完整的结果列表，而且这个列表的顺序和 header 里的签名顺序一致。如果我需要在所有店铺计算完成后发一条通知或者落一个汇总表，这就是最干净的方案。

5.3 chunks：大批量任务的分片处理思路

回到第 4 节的场景，当任务数超过几千甚至上万时，chunks 几乎就是标配。它和 group 最大的区别在于：group 把每个任务都作为一个独立消息塞进 broker，chunks 把多个任务打包成一个分片消息，broker 压力指数级下降。

我从 5 万任务的实践里总结出的经验是：单批任务 500 到 1000 之间比较合适。太小的分片（比如 10 个）没有明显减负效果，太大的分片（比如 5000 个）又可能造成 worker 单次处理时间过长，分片之间负载不均。

5.4 组合实例：签名流 + 分组 + 聚合

顺手给一个相对完整的多步骤场景，方便你直接参考。

需求是：每天都有一批订单文件需要处理，先解析文件，然后按渠道分组做并发计算，最后把每个渠道的结果统一入库。

from celery import chord, group from tasks import parse_file, calc_channel, save_channel_result def daily_report(file_list): # 每个文件先解析，再对每个文件内部的数据做并发计算 tasks = [] for file in file_list: tasks.append( chain( parse_file.s(file), chord( group(calc_channel.s(file, ch) for ch in channel_list), save_channel_result.s(file) ) ) ) # 所有文件处理完成后可以再触发一个全局通知 return group(tasks).apply_async()

这个例子里，外层用 group 让多个文件并行处理，内层用 chain 保证“解析 -> 计算 -> 入库”的顺序，用 chord 保证“所有渠道算完”再入库。canvas 的编排思路在这里就体现得很完整了：它不只是任务的集合，而是任务之间关系的描述语言。

我用 Celery canvas 这一年多最大的体会是：group 看似只是一个“并发分发”的工具，可一旦你开始用“签名”而不是“直接调用”的视角看待任务，整个系统的可编排性就完全不一样了。批量任务不再是堆代码，而是像搭积木一样，把一个个签名合理地组合起来。最后再分享两个小建议：第一，所有任务函数都要有明确的超时和异常设计，这比用什么画布编排都重要；第二，先小批量验证结果顺序和失败行为，再上大规模，canvas 的原语虽好，但每一步推演都要心里有数。