做后端时间久了你会发现任务队列这事儿难的不是定义任务而是怎么编排任务。我用 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(timeout10) # 拿到数据再去整合这个写法能用吗能。但它有两个问题在真实项目里非常要命。第一是代码层面散。任务的发送逻辑和结果收集逻辑被拆散到不同的函数里一旦中间夹了其他业务逻辑后面的人读代码根本看不出来这 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(timeout30)超时控制可以在组级别统一配置第三作为编排单元与其他原语组合因为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(x1, y2) # 关键字参数 # 也可以部分绑定参数后面再补 sig3 add.s(1) # 先绑一个 sig3.args (2,) # 或者合并签名时再补这个灵活性很重要。实际业务里经常遇到“统一参数 各自差异参数”的组合签名机制能让你把公共参数提前绑定好差异参数在生成 group 时动态补齐。2.2 group 的返回值apply_async 之后你拿到的是什么group.apply_async()返回的是一个GroupResult对象。别看它名字里带 Group它本质上是对多个 AsyncResult 的封装内部维护着一个任务 ID 列表。这个对象有几个关键属性和方法属性/方法作用使用频率.children包含所有子任务 AsyncResult 的列表中.results子任务 ID 与 AsyncResult 的映射或列表取决于后端中.get()阻塞等待并返回所有子任务的结果列表极高.join()与 get() 类似但更强调“等待完成”的语义高.save()保存 GroupResult 到后端便于跨进程恢复低.forget()清除所有子任务的结果数据低.revoke()终止组内的所有任务中通常我习惯直接调用result.get(timeout10)。如果里面某个任务失败了默认propagateTrue那么get()会抛出这个任务对应的异常如果设置propagateFalse则失败任务的位置会是异常对象其他任务结果照常返回。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(timeout30) # 直接 zip 使用 for shop_id, res in zip(shop_id_list, res_list): ...这个有序性是收官收得非常舒服尤其做报表拼接、批量导出这类场景少写一大堆“手动匹配”的代码。2.4 参数细节timeout、propagate、intervalgroup 级的get()/join()有一些细节参数平时文档里写得很简略实战中却直接决定着你的程序是 3 秒跑完还是 3 分钟卡死。先说timeout。这是指等待所有子任务完成的最长秒数到点还没完成就抛TimeoutError。注意它不代表子任务会被终止子任务可能还在 worker 上跑。如果你在 Web 请求里使用result.get()一定要设置合理的 timeout否则这个请求会一直挂着连接池迟早被耗尽。再说interval。get()底层并不是一次性阻塞到结果出来而是每隔一段时间去后端轮询一次。默认的interval0.5也就是每 0.5 秒检查一次。如果你有 500 个任务执行时间整体约 2 秒这个默认值完全够。但如果你用的是非默认的结果后端轮询频率和网络开销之间就得权衡一下。最后是propagate。在生产环境我建议大多数场景把它设为 False然后手动在结果列表里过滤异常。原因很简单一个 100 任务的 group可能只有 1 个任务因为脏数据失败了如果直接propagateTrue抛异常剩下 99 个正常结果你也拿不到还得额外重查非常浪费。res_list result.get(timeout30, propagateFalse) 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, brokerredis://localhost:6379/0, backendredis://localhost:6379/1, ) app.conf.update( task_serializerjson, result_serializerjson, accept_content[json], timezoneAsia/Shanghai, enable_utcTrue, task_track_startedTrue, ) app.task(bindTrue, max_retries3, default_retry_delay5) 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(fshop {shop_id} data inconsistent) return {shop_id: shop_id, total: round(shop_id * 3.14 * factor, 2)}代码里的bindTrue可以把任务实例绑定进来重试时用self.retry()。max_retries3这个参数配合 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, factor1.0): # 构造签名列表公共参数 factor 提前绑定差异参数 shop_id 逐个传入 sigs [calculate_shop_summary.s(shop_id, factorfactor) for shop_id in shop_ids] # 创建 group 对象并异步提交 group_result group(sigs).apply_async() print(fGroup Task ID: {group_result.id}) print(fChild 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, factor1.2) # 统一收集结果设置 propagateFalse 避免单个失败影响整体 res_list result.get(timeout60, propagateFalse) elapsed time.perf_counter() - start print(fBatch completed in {elapsed:.2f}s) print(fSuccess: {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 运行与验证并发效果和结果收集先启动 workercelery -A tasks worker --loglevelinfo --poolprefork --concurrency8然后运行主程序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_id17: ValueError(shop 17 data inconsistent) Failed shop_id34: ValueError(shop 34 data inconsistent)这里有个很直观的验证点30 个任务每个任务time.sleep(2)如果串行要 60 秒。但因为并发度是 8整体耗时约 9.4 秒说明 group 确实是并发执行的。这就是为什么批量任务用 group 而不是 for 循环顺序发送的底气所在。3.4 为什么这个 Demo 的设计值得借鉴这个 Demo 虽然简单但几个设计习惯是经过生产检验的。第一公共参数提前绑定差异参数最后补齐。factor1.2在签名构造时统一传入shop_id则在列表推导式里逐个传入。这样后面如果要在不同业务模块里复用改动成本极低。第二propagateFalse 手动过滤失败。假设这里用默认的 propagateTrue一个店铺数据异常整个组的 get() 直接抛异常主流程就得中断其他 29 个正常结果全部丢失。手动过滤后失败的可以进入重试队列成功的直接落库容错性明显更好。第三在任务内部做重试。max_retries3是针对偶发抖动、网络闪断这类异常的兜底而业务脏数据异常并不适合在任务内部死磕因为重试 3 次也一样会失败。所以任务的异常设计要分清“可重试”和“不可重试”不可重试的尽早抛给上层处理。4. 生产环境中踩过的 group 坑从现象到根因的排查链路4.1 坑一任务偶发丢失结果列表少了几项怎么办某次压测时发现100 个任务发送出去最后get()到的结果只有 97 个且没有超时。这个现象非常让人抓狂因为任务队列看起来一切正常worker 日志里也没有报错。排查链路是这样的先看 worker 日志发现确实只处理了 97 个任务。接着看 Redis 里的队列长度发送后立刻查询发现消息入队 100 条。然后看 result 后端发现那 3 个丢失任务对应的结果键从未被写入。再回头检查任务本身发现其中 3 个任务抛了异常但异常被配置项task_acks_on_failureFalse且 worker 默认不把失败任务标记为“已确认”导致消息不断重试不是仔细看日志发现异常直接中断了任务执行但 worker 进程没有崩溃。最终根因是任务函数里某段代码依赖外部接口偶发超时抛错而异常发生后 worker 默认行为是“确认消息并记录失败”。但那个环境里配置了task_acks_lateTrue消息并没有被 ACK同时任务也没有写失败结果到 backend于是消息被重新投递但同样的错误再次发生。看起来就像是“结果丢失”。解决方式分两层业务上给外部依赖调用加重试框架上把失败结果写回后端防止排查时误判为丢失。排查过程中反复对比任务日志、消息队列、结果后端三方数据抓到了根因。技巧遇到 group 结果不齐第一时间对比“worker 日志完成数”、“队列消费数”、“结果后端写入数”三个数字。三者不一致问题大概率出现在任务执行或 ACK 配置上。4.2 坑二大 group 导致 broker 压力陡增发送端的处理有一次批量业务上线单批任务数量直接上到 5 万。group 对象倒是构造出来了但apply_async发送时Redis broker 的 CPU 瞬间飙到 90%消息积压严重其他业务的延迟跟着遭殃。原因很好理解group 本身不会“一次性发送”它会逐个把子任务发到 broker5 万个任务一条条入队瞬间写入压力全部压在 Redis 上。解法不是不用 group而是给 group 加“限流”。Celery canvas 提供了chunks它是 group 的“分片版”每片指定大小逐个发送。比如每 500 个签名为一个 chunk5 万任务就变成 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 分钟。主调用方设置的timeout30于是get()抛了超时。问题在于超时任务并没有终止它还在 worker 上继续跑结果就是主流程已经判定失败worker 还在空转消耗资源。这个问题的根因在于 group 的超时控制。官方文档没有把话说透group 的 timeout 只是等待超时并不负责杀掉超时子任务。正确做法是在设计任务时就要考虑“任务必须能自然结束”。所有外部调用都要设置 socket 超时和业务超时杜绝任务永远悬挂。对于确实无法预估时长的任务可以在组级超时后主动revoke这些子任务。try: result group_result.get(timeout30) except TimeoutError: # 收集超过 30 秒仍未完成的任务 pending_results [r for r in group_result.children if not r.ready()] for r in pending_results: r.revoke(terminateTrue, signalSIGKILL) raise这里revoke(terminateTrue)会强制杀掉对应 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 把每个任务都作为一个独立消息塞进 brokerchunks 把多个任务打包成一个分片消息broker 压力指数级下降。我从 5 万任务的实践里总结出的经验是单批任务 500 到 1000 之间比较合适。太小的分片比如 10 个没有明显减负效果太大的分片比如 5000 个又可能造成 worker 单次处理时间过长分片之间负载不均。任务规模推荐方案原因 100普通 group简单直接结果收集方便100 ~ 5000group 合理并发worker 并发度足够覆盖 5000chunks 分片降低 broker 写入压力分片消费更平滑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 的原语虽好但每一步推演都要心里有数。