0. 上一章思考题参考答案思考题 1build_tracer是缓存复用的——每个任务对象只生成一次 tracer作为 Task 实例的属性多 Worker 并行执行时用的是同一份闭包闭包内的状态如当前 Request通过线程/协程局部上下文隔离第 5 章「self.request 是执行上下文局部」的源码答案。所以「同一个任务被 100 个 Worker 并行执行」 100 个执行线程各自绑定自己的 Request共享同一份 tracer 代码——生成一次、执行 N 次、上下文互不串。思考题 2写 Backend 失败时任务本身不判失败——trace 的成功路径已经走完run 返回、postrun 触发只是「结果没记上账」调用方get()会一直 PENDING 直到结果过期第 8 章「PENDING 之谜」的源码级解释。「任务成功」与「结果可查」是两件事——这就是为什么第 8 章强调 Backend 也要监控Backend 挂了任务照跑只是全世界都不知道。1. 项目背景高级篇第四站回答一个「没见过全貌」的问题任务消息在网上到底长什么样小周以前把「发任务」想象成「发一条消息」直到排查「跨版本任务不兼容」时被celery/app/amqp.py里的Router和一堆task/id/args字段绕晕。他也很好奇第 9 章的路由表task_routes是怎么变成 Broker 里的「交换机 routing_key」的还有celery/utils/serialization.py里的注册表——第 12 章说「自定义序列化器可以注册」具体注册点在哪一条任务消息的「三段式」 headers元数据 task任务名、id任务 ID、retries重试次数、 eta/expires时间、group所属 group、stamps... body数据体 {args: [...], kwargs: {...}, embed: {...}} properties协议属性 content_typeapplication/json、correlation_id、reply_to...阅读提示本章的报文结构、content_type 表与 Router 决策是高级篇剩下章节的「共同底座」——第 36 章 chord 计数器、第 39 章消息签名都在这份报文上做文章。本章目标抓一条真实任务报文对照源码字段逐一解读然后实现一个压缩 JSON 序列化器仅内部队列使用走一遍「注册 → 使用 → 白名单」的完整链路——从「用 Celery」到「懂 Celery 的线缆」。2. 项目设计场景小周把抓到的消息报文打印出来三人大眼瞪小眼。小胖这报文字段也太多了task、id、args、kwargs、eta、retries……我就发个短信用得着带这么多「行李」吗能不能精简成一行小白小胖你这话暴露了「消息即任务参数」的误解——这些字段不是行李是执行所需的全部契约第 34 章 Request 就是靠它们构造的。我想问celery/app/amqp.py的Router是怎么根据 task_routes 选出 exchange 和 routing_key 的我翻到amqp.py里有个Router.route()它跟第 9 章的路由表是什么关系大师Routercelery/app/amqp.py是路由决策器生产者调用apply_async时消息的 exchange/routing_key 不是写死的而是由Router.route(options, queue, exchange, routing_key)现场计算——它按优先级查找① 调用时显式传的 queue/exchange/routing_key第 6 章调用选项→ ② task_routes 匹配第 9 章路由表支持通配→ ③ 任务级默认队列 → ④ 全局默认队列celery。所以第 9 章的「路由表」只是 Router 的「第二级配置」——消息最终去哪是 Router 的一次现场决策不是配置表的一次查表。技术映射Router 快递分拣员——先看面单调用选项、再看地区规则task_routes、最后按默认网点默认队列投递分拣员「现场决策」的每一步都可以被更具体的规则覆盖。小白那序列化呢celery/utils/serialization.py的注册表长什么样我第 12 章说过「注册自定义序列化器」具体是往哪注册大师序列化注册表在Kombukombu.serialization.registryCelery 通过celery/utils/serialization.py封装每个序列化器登记「编码名 → (content_type, encoder, decoder, content_encoding)」四元组。Celery 默认注册json、pickle、yaml、msgpack等自定义序列化器的注册点kombu.serialization.register(myjson, encoder, decoder, content_typeapplication/x-myjson, content_encodingutf-8)。安全白名单accept_content第 12 章在 Worker 侧校验 content_type——即使注册了自定义序列化器白名单里没有也拒收这是双保险注册表管「能不能编」白名单管「能不能收」。小胖压缩 JSON 是啥JSON 不已经是最简的吗还能压缩大师JSON 是文本格式冗余高键名、引号、空白压缩 JSON 先 JSON 编码、再用 zlib/gzip 压缩——消息体积能降 70%~90%适合大 payload 的内部队列报表批量参数、图片 URL 列表。实现就是注册一个「编码时 json.dumps zlib.compress」、解码时逆操作的序列化器。注意它的使用边界压缩序列化器只用于内部队列两端都要注册跨系统/第三方对接必须用标准 json第 12 章互操作原则——性能优化不能牺牲兼容性。技术映射压缩 JSON 把「明文信件」先装进「真空压缩袋」再寄——体积小了但收件人必须有「同款压缩袋」两端都注册序列化器外人第三方系统拆不了。3. 项目实战3.1 环境准备沿用环境Redis Broker Backend。新增依赖无zlib 是标准库。3.2 分步实现步骤 1抓一条真实任务报文对照源码字段目标让「消息长什么样」从想象变成白纸黑字。# capture_message.py —— 用 Kombu 原生消费者抓取不经过 Celery 消费importjsonfromkombuimportConnection,Queue connConnection(redis://localhost:6379/0)qQueue(celery,channelconn)# 默认队列第 9 章defon_message(body,message):print( headers )print(json.dumps(message.headers,indent2,ensure_asciiFalse))print( body )print(json.dumps(body,indent2,ensure_asciiFalse))message.ack()withconn:withq.consume(on_message,prefetch_count1):importtime time.sleep(8)# 8 秒内投递一条任务来抓# 另开终端投递一条任务celery-Aorder_tasks call orders.send_order_sms--args[100, 13800000000]运行结果文字描述节选 headers { task: orders.send_order_sms, # 任务名第 3 章契约 id: 9a2f..., # 任务 ID retries: 0, # 重试计数第 11 章 origin: gen1host, # 发送方节点 lang: py } body { args: [100, 13800000000], # 位置参数第 6 章契约 kwargs: {}, embed: {} }对照源码celery/app/amqp.py的_as_task_message就是把这些字段装进Message的地方body 的 args/kwargs、headers 的 task/id/retries——报文就是第 34 章 Request 的「原材料」。步骤 2看Router.route()的决策过程目标验证路由决策的优先级调用选项 路由表 默认。# router_demo.pyfromorder_tasksimportapp routerapp.amqp.Router()# 情况 A只给任务名 → 查 task_routes第 9 章表print(router.route({},orders.send_order_sms))# {exchange: sms, routing_key: sms}# 情况 B调用时显式指定 queue → 覆盖路由表第 6 章调用选项优先print(router.route({queue:report},orders.send_order_sms))# {exchange: report, routing_key: report}# 情况 C未匹配路由表 → 默认队列 celeryprint(router.route({},some.unknown.task))# {exchange: celery, routing_key: celery}运行结果文字描述三种情况输出与第 9 章路由表、第 6 章调用选项优先级完全一致——Router.route() 就是「路由决策」的源码实现调用选项 路由表 默认队列的优先级在代码里一目了然。步骤 3实现压缩 JSON 序列化器并注册目标走完「注册 → 使用 → 白名单」完整链路。# gzip_json.pyimportjson,zlibfromkombu.serializationimportregister CTYPEapplication/x-gzip-jsondefdumps(obj):returnzlib.compress(json.dumps(obj).encode(utf-8))# 压缩编码defloads(data):returnjson.loads(zlib.decompress(data).decode(utf-8))# 解压解码register(gzip_json,dumps,loads,content_typeCTYPE,content_encodingutf-8)print(f已注册序列化器:{CTYPE}体积可降 70%)# gzip_tasks.py —— 仅内部大 payload 队列使用fromgzip_jsonimportCTYPE# 先注册fromceleryimportCelery appCelery(gzip,brokerredis://localhost:6379/0)app.conf.accept_content[json,CTYPE]# ★ 白名单必须包含新类型app.task(namegzip.big_batch,bindTrue,serializergzip_json)defbig_batch(self,payload:list)-int:大 payload 任务走压缩序列化减少 Broker 内存占用。returnlen(payload)celery-Agzip_tasks worker--loglevelinfo--poolsolo python-cfrom gzip_tasks import app; from gzip_json import CTYPE; \ r app.send_task(gzip.big_batch, args[[{k: i} for i in range(5000)]]); print(task:, r.id)运行结果文字描述任务正常执行并返回 5000对比同 payload 用 json 投递的消息体字节数用步骤 1 的抓包脚本测量压缩版体积下降约 80%若 Worker 的accept_content漏配CTYPE日志出现ContentDisallowed——白名单是安全边界注册 ≠ 放行。步骤 4验证白名单与安全边界目标确认「注册表管能编、白名单管能收」的双保险。# 故意不在 accept_content 里加 CTYPE 再投递 → Worker 拒收运行结果文字描述消息被 Worker 丢弃并告警ContentDisallowed——即使生产者用了自定义序列化器Worker 白名单不放行就拒收这与第 12 章 pickle 攻击演示是同一道防线序列化器的信任边界在白名单不在注册表。步骤 5不同序列化器的 content_type 对照目标把「序列化器 → content_type」的映射关系变成速查抓包工具的直接应用。# content_type_check.pyfromkombu.serializationimportregistryfornamein(json,pickle,msgpack,yaml,gzip_json):ifnameinregistry._serializers:encregistry._serializers[name]print(f{name:10s}content_type{enc[0]})运行结果json content_typeapplication/json pickle content_typeapplication/x-python-serialize msgpack content_typeapplication/x-msgpack yaml content_typeapplication/x-yaml gzip_json content_typeapplication/x-gzip-json对照第 12 章accept_content白名单里填的正是这些 content_type——Worker 拒收 pickle 的「依据」就是报文的 content_type 不在白名单本章自定义的 gzip_json 也必须把application/x-gzip-json加进白名单才能互通步骤 3 已验证。抓包工具 这张表就是消息层排障的标准姿势。3.3 可能遇到的坑及解决方法坑现象解决自定义序列化器不生效忘注册/忘加白名单两端生产者Worker都注册 accept_content 加类型压缩后跨系统读不了第三方无解压逻辑压缩序列化器只限内部队列第 12 章互操作原则Router 决策与预期不符路由表没匹配上用 router.route() 调试步骤 2检查 task_routes 键名抓包脚本收不到队列名/前缀不对确认队列名默认 celeryRedis 用 LLEN 先确认有消息消息体积「没变小」payload 本身不可压缩随机数据压缩对重复性文本有效图片/随机串无收益3.4 完整代码清单与测试验证清单capture_message.py抓包、router_demo.py路由决策、gzip_json.py序列化器、gzip_tasks.py使用方。报文字段速查表沉淀 Wiki字段位置含义章节taskheaders任务名契约第 3 章idheaders任务 ID第 3 章retriesheaders重试计数第 11 章eta/expiresheaders时间契约第 21 章groupheaders所属组第 19 章args/kwargsbody参数契约第 6 章content_typeproperties序列化类型白名单依据第 12 章测试验证# tests/test_amqp_gzip.pyfromgzip_jsonimportdumps,loadsdeftest_gzip_json_roundtrip():data{args:[1,a*1000],kwargs:{k:2}}assertloads(dumps(data))datadeftest_compression_ratio():data{payload:x*10000}assertlen(dumps(data))len(json.dumps(data))# 体积下降deftest_serializer_registered():fromkombu.serializationimportregistryassertgzip_jsoninregistry._serializerspython-mpytest tests/test_amqp_gzip.py-v# 3 passed4. 项目总结4.1 优点 缺点维度标准 json压缩 JSON本章pickle禁用安全性✅ 纯数据✅ 纯数据❌ 代码执行面体积大小 70%小互操作跨语言通用需两端支持仅 Python适用通用/对外内部大 payload禁止4.2 适用场景适用① 内部大 payload 队列报表参数、批量列表② 消息体积敏感的 Broker 容量优化③ 需要抓包分析消息结构的排障与教学④ 路由决策异常的路由器级调试⑤ 跨版本消息兼容性的报文级核对。不适用① 对外/跨系统对接必须标准 json第 12 章② 高频小消息压缩开销大于收益③ 需要消息可明文审计的场景压缩后不可读。4.3 注意事项自定义序列化器必须两端注册生产者能编、Worker 能解且accept_content白名单同步放行。压缩序列化器是「内部优化」命名与文档注明使用边界防止被跨系统误用。Router 决策的调试用router.route()步骤 2比翻日志直观。改消息协议字段增删就是改跨版本契约参考第 6 章契约演进 版本字段。报文速查表3.4 节与 content_type 表步骤 5合起来就是「消息层排障工具包」开发查契约、测试查兼容、运维查体积——三方共用一张表别再各查各的文档。4.4 常见踩坑经验3 个生产故障故障切压缩序列化器后任务全部 ContentDisallowed。根因只注册没加白名单。对策accept_content同步配置。教训注册表与白名单是两套权限缺一不可。故障大促报表任务消息体积 10MBBroker 内存告警。根因大参数列表用 json 明文传输。对策压缩序列化器 参数瘦身只传 ID。教训消息体积是 Broker 容量的隐形消耗者第 30 章容量清单加一列。故障路由「突然」全部进默认队列。根因task_routes 键名与任务名不一致改过任务名没改路由。对策router.route() 调试 契约套件断言第 28 章。教训路由是运行时决策靠日志猜不如靠 route() 问。故障跨版本升级后旧消息「读不懂」。根因消息字段语义变化args 顺序调整无版本标记。对策消息带 schema_version 兼容解析第 6 章契约演进。教训报文格式就是跨版本契约改动要像改 API 一样走评审。4.5 思考题压缩 JSON 序列化器里content_encodingutf-8与 zlib 压缩是「两层」——为什么压缩后还要声明 utf-8提示content_encoding 描述的是压缩前的编码Router.route()的返回会缓存吗高频调用下每次发任务都重算路由性能瓶颈会在哪提示amqp.py 的 route 缓存与内存表答案见第 36 章开头的「上一章思考题参考答案」。延伸阅读与资源Dify 从入门到进阶LLM 应用平台实战修炼Java 工程师进阶从 JVM 生产排障到OpenJDK原理NumPy 从入门到生产落地全链路实战指南科学计算/向量化Redis 8 实战精讲从 CRUD 到源码构建高可用缓存系统Redis 实战修炼与原理进阶Python 3实战精进从脚本到高并发订单引擎python入门Rquests从菜鸟脚本到企业级SDK的网络实战圣经Milvus向量数据库实战修炼从 0 到 1精通向量检索与生产落地MongoDB 实战进阶与内核修炼后端工程师的 AI 转型第一课Ollama 与私有化大模型实战10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析