0. 上一章思考题参考答案思考题 1撤销集合是时序敏感的——「消费前必须知道哪些不能动」Mingle 在 Tasks 前因为消费动作会立刻执行任务而 Worker 状态谁活着是最终一致即可——晚几秒知道邻居挂了不影响正确性Gossip 在 Tasks 后边跑边传。顺序设计的原则先同步「会阻止动作的记忆」后传播「不影响动作的状态」。思考题 2TasksStepcelery/worker/consumer/tasks.py在声明消费者后通过task_consumer.qos(prefetch_countN)设置 QoS——N 并发 ×worker_prefetch_multiplier第 18 章公式在消费者声明时生效由 kombu 底层转成 AMQP 的 basic.qos 帧Redis transport 转成可见性窗口语义。所以「预取数」是 Broker 侧的消费约束不是 Worker 侧的逻辑。1. 项目背景高级篇第三站也是最硬核的一站任务执行引擎。第 5 章我们用过self.request、第 11 章用过self.retry()、第 26 章挂过task_prerun信号——但「任务函数被调用前后内核到底做了什么」一直没有全景。小周的困惑来自一次「慢任务排查」短信任务日志显示succeeded in 2.003s但任务函数体里明明只 sleep 了 1 秒——另外 1 秒去哪了大师提示他去看celery/app/trace.py和celery/worker/request.py——任务执行不是一个「函数调用」而是「一条流水线」消息 → Request反序列化 上下文→ strategy投递策略→ build_tracer生成执行器 → trace_taskprerun → run → 成功/失败/重试 → postrun → 写 Backend那 1 秒的「消失时间」就藏在流水线里反序列化消息体 JSON 解析、Request 构造、tracer 的各环节钩子、Backend 写入——它们都是「任务耗时」的一部分但任务函数体感知不到。本章目标读懂这条流水线并给trace_task增加一个慢任务火焰日志执行 3s 打印当前栈用真实短信任务验证——既解答「时间去哪了」又拿到一个生产可用的排查工具。一句话先记住本章结论「发任务」在消息层完成「执行任务」在 trace 层完成——第 34 章是中间那段「从消息到函数体」的全部秘密。2. 项目设计场景小周把「2 秒 vs 1 秒」的差异摆在桌面大师开始拆流水线。小胖2 秒对 1 秒不就差 1 秒嘛sleep 多了呗有啥好查的你们搞源码的净纠结这些细节小白小胖你这个「多了呗」就是「拍脑袋排障」——第 15 章批过无数次了。我看了下celery/worker/request.py有一个Request类还有celery/worker/strategy.py的default函数。我想先问消息进来后Request 是怎么被构造的strategy 又管什么大师顺序是消息到达 → Tasks Step 把消息交给 strategy投递策略→ strategy 决定「何时执行、怎么执行」→ 交给 Request 构造执行上下文 → 进入 trace。Requestcelery/worker/request.py的职责是把消息反序列化成任务执行所需的上下文task_id、args/kwargs、headers、retries、delivery_info队列/交换机——它就是第 5 章self.request背后的那个对象。strategycelery/worker/strategy.py是调度决策层处理countdown/eta放 Timer第 21 章、rate_limit限速节奏、acks_late的确认时机——「什么时候执行」由 strategy 决定「执行成什么样」由 trace 决定。技术映射strategy 前台接待决定「这单什么时候进后厨」延时单放 Timer 冷藏、限速单排队Request 点菜单把顾客的话整理成后厨能懂的规格菜名、忌口、桌号trace 后厨炒菜流水线备菜→炒→出锅→上账。小白那build_tracer和trace_task呢我看了celery/app/trace.py——build_tracer:344看起来是「生成执行器」的工厂trace_task:689是「入口」。它们的生命周期到底是啥大师build_tracer是工厂根据任务对象生成一个「带完整钩子的执行函数」——它把信号第 26 章、重试逻辑第 11 章、Backend 写入第 8 章、异常捕获全部编织进一个闭包trace_task是入口拿到消息后调用这个 tracer。完整的执行链源码对照trace_task:689 └─ build_tracer 生成 tracer:344 ├─ task_prerun 信号第 26 章 ← 第 5 章 self.request 就绪 ├─ 调用 task.run(*args, **kwargs) ← 任务函数体1 秒 sleep ├─ 成功 → task_postrun → 写 Backend(SUCCESS) 结果 ├─ 失败 → task_failure 信号 → 判定「重试 or 终态」 │ ├─ 可重试 → task.retry() 抛 Retry → 状态 RETRY → 重投 │ └─ 终态 → 写 Backend(FAILURE) traceback └─ 返回 (state, retval, ...)那 1 秒的「消失时间」 信号钩子 反序列化 Backend 写入 结果序列化的总开销——任务函数体只是流水线的一段不是全部。小胖那异常呢任务里抛了个ValueError谁会捕获捕获之后 ExceptionInfo 又是啥听着像「异常的档案袋」大师好比喻。trace 的异常处理链task.run抛异常 → tracer 捕获 → 包装成ExceptionInfocelery/app/trace.py的 ExceptionInfo 类持有异常对象 traceback 文本 可序列化→ 分派若匹配任务的重试条件第 11 章 autoretry/手动 retry→ 走 Retry 分支任务状态 RETRY消息重投否则 →task_failure信号第 26 章死信入口→ 写 Backend FAILUREget()时 propagate 回调用方第 8 章。ExceptionInfo 的价值异常可以被序列化跨进程传递——Worker 进程的异常能原样出现在调用方get()的堆栈里靠的就是它。技术映射ExceptionInfo 事故的「档案袋」——现场照片traceback 当事人异常对象装袋归档袋子能跨部门进程寄送调用方拆袋就能看到事故全貌。3. 项目实战3.1 环境准备沿用环境Redis Broker Backend。本章通过给 trace 增加慢任务火焰日志源码级实验可编辑安装验证执行链路的每一段耗时。3.2 分步实现步骤 1给trace_task增加慢任务火焰日志目标任务执行 3 秒时打印当前栈与各阶段耗时。# slow_trace.py —— 通过信号实现不改框架源码更安全的生产方案# 但本章目标是「读懂 trace」先做一个源码级插入importtimefromceleryimportsignals# 方案 A生产推荐信号采集各阶段耗时第 26 章_stage_t0{}signals.task_prerun.connectdeft0(sender,task_id,task,**kw):_stage_t0[task_id]time.time()signals.task_postrun.connectdeft1(sender,task_id,task,retval,state,**kw):t_start_stage_t0.pop(task_id,None)ift_startand(time.time()-t_start)3:print(f[SLOW]{task.name}{task_id}耗时{time.time()-t_start:.2f}s)importtraceback traceback.print_stack()# 打印调用栈慢在哪一段celery-Aorder_tasks worker--loglevelinfo--poolsolo-Qsms# 投递一个 4 秒的慢任务sleep 4celery-Aorder_tasks call orders.slow_task--args[4]--queuesms运行结果文字描述任务完成后打印[SLOW] orders.slow_task ... 耗时 4.02s 调用栈——慢任务被自动标记栈里能看到执行路径trace_task → tracer → task.run → sleep。步骤 2拆解「任务耗时」的各阶段回答 2s vs 1s目标量化流水线各段开销看清「消失的时间」。# stage_timing.py —— 各阶段计时信号方案importtimefromceleryimportsignals _ctx{}signals.task_prerun.connectdefpre(sender,task_id,task,**kw):_ctx[task_id]{t0:time.time()}signals.task_postrun.connectdefpost(sender,task_id,task,retval,state,**kw):c_ctx.pop(task_id,{})c[postrun]time.time()# 这里只能看到 prerun→postrun 的框架开销函数体内耗时由任务自己统计print(f[STAGE]{task.name}: 框架开销 ≈{c[postrun]-c[t0]:.3f}s)运行结果文字描述短任务函数体 1s的框架开销约 5~30ms信号Backend 写入长任务 2s vs 1s 的差异主要在「排队等待 预取 反序列化」不在流水线钩子——结合inspect scheduled/active第 15 章就能定位「时间去哪了」。步骤 3验证异常处理链——Retry 与 ExceptionInfo目标读trace_task的分支验证重试与终态的判定路径。# retry_flow_demo.pyfromorder_tasksimportappapp.task(nameorders.demo_retry,bindTrue,max_retries2,autoretry_for(ValueError,))defdemo_retry(self,flag:str)-str:ifflagbad:raiseValueError(业务错误应被自动重试)returnokcelery-Aorder_tasks worker--logleveldebug--poolsolo-Qorder celery-Aorder_tasks call orders.demo_retry--args[bad]# 观察 Worker 日志与结果celery-Aorder_tasks resulttask_id运行结果文字描述节选[DEBUG] Task orders.demo_retry[..] retry: Retry in 0s # 异常 → Retry 分支 [DEBUG] Task orders.demo_retry[..] succeeded in ... # 重试后成功 # 若把 max_retries0 重跑 [ERROR] Task orders.demo_retry[..] raised ValueError # 终态分支 [INFO] Task orders.demo_retry[..] FAILURE # 写 Backend FAILURE对照 trace 源码retry: Retry in Ns对应build_tracer里捕获Retry异常的路径raised ... FAILURE对应终态写 Backend 路径——源码里的每个分支都在日志里有对应痕迹。步骤 4慢任务火焰日志的生产化收敛为工具目标把实验代码收敛成「生产可用的慢任务告警器」。# slow_task_monitor.py —— 生产版信号方案第 26 章规范只记账不打栈importtimefromprometheus_clientimportCounterfromceleryimportsignals SLOWCounter(celery_slow_task_total,慢任务计数3s,[task])_ctx{}signals.task_prerun.connectdef_t0(sender,task_id,task,**kw):_ctx[task_id]time.time()signals.task_postrun.connectdef_t1(sender,task_id,task,retval,state,**kw):t0_ctx.pop(task_id,None)ift0and(time.time()-t0)3:SLOW.labels(task.name).inc()# 只计数慢任务定位交给日志/追踪print(f[SLOW]{task.name}耗时{time.time()-t0:.2f}stask_id{task_id})运行结果文字描述与第 25 章 Prometheus 体系对接——celery_slow_task_total指标进大盘慢任务数突增 告警「慢任务火焰日志」从实验工具升级为生产监控组件排查时再用 traceback.print_stack 的调试版。步骤 5子任务上下文栈——任务里发任务的「栈」目标理解「任务 A 里调用任务 B」时trace 的上下文如何嵌套第 3 章「子任务」的源码层。# context_demo.pyfromorder_tasksimportappapp.task(nameorders.parent,bindTrue)defparent(self,child_count:int)-str:print(f[parent] 当前栈深:{len(self.request.children)ifhasattr(self.request,children)elsen/a})foriinrange(child_count):child.delay(i)# 任务里发任务第 3/6 章returnparent-doneapp.task(nameorders.child,bindTrue)defchild(self,idx:int)-str:returnfchild-{idx}celery-Aorder_tasks worker--loglevelinfo--poolsolo-Qorder celery-Aorder_tasks call orders.parent--args[3]运行结果文字描述parent 执行后child任务被投递 3 次self.request里能通过children等字段看到子任务关联——trace 的执行上下文是「栈式」的每个任务执行时压入当前 Request执行完弹出第 5 章self.request线程局部的源码机理。生产里「子任务结果汇总回父任务」就是靠这层栈 第 23 章结果树实现的。3.3 可能遇到的坑及解决方法坑现象解决慢任务不触发日志阈值 实际耗时阈值按 P99 校准第 30 章先观察再定值信号里 print 刷屏每个任务都打只打超阈值或用指标计数步骤 4Retry 分支看不到日志级别不够--logleveldebug看 retry/raised 痕迹ExceptionInfo 序列化失败异常对象不可 pickle业务异常继承标准 Exceptiontraceback 文本兜底框架开销误判为慢prerun→postrun 差 30ms 以为是框架问题区分「队列等待 预取」与「执行流水线」两段3.4 完整代码清单与测试验证清单slow_task_monitor.py生产版、stage_timing.py阶段计时、retry_flow_demo.py异常链验证。trace 生命周期速查沉淀 Wiki对照celery/app/trace.py阶段源码位置触发点反序列化Requestworker/request.py消息到达投递决策strategyworker/strategy.pyETA/限速/ack 时机执行器生成build_tracertrace.py:344首次执行任务执行trace_tasktrace.py:689prerun→run→postrun重试分支Retry 异常捕获匹配重试条件终态写回Backend store_resultSUCCESS/FAILURE测试验证# tests/test_trace.pyfromslow_task_monitorimport_ctxfromorder_tasksimportapp,demo_retry app.conf.task_always_eagerTruedeftest_retry_task_defined():fromorder_tasksimportdemo_retryassertdemo_retry.max_retries2assertValueErrorindemo_retry.autoretry_fordeftest_exception_goes_failure_when_no_retry():app.task(nameorders.no_retry)defno_retry():raiseValueError(boom)rno_retry.apply()assertr.failed()deftest_slow_monitor_ctx_clean_after_postrun():# 正常执行后 _ctx 应清空无泄漏fromorder_tasksimportsend_order_sms send_order_sms.apply(args[1])assertlen(_ctx)0python-mpytest tests/test_trace.py-v# 3 passed4. 项目总结4.1 优点 缺点维度信号方案不侵入改 trace 源码方案安全性零侵入生产可用改框架需还原覆盖钩子级信号点全链路任意插入维护独立文件与版本耦合调试深度看得到钩子看得到全部推荐生产监控学习/实验4.2 适用场景适用① 慢任务定位火焰日志/告警② 理解「任务耗时」构成排障、容量计算第 30 章③ 异常链路的可观测重试/终态判定④ 生产级慢任务监控指标 告警⑤ 子任务上下文与结果树的源码理解第 23 章血缘的延伸。不适用① 业务任务逻辑trace 是框架层业务别碰② 需要「修改异常处理语义」的场景改 trace 风险大优先用信号或自定义 Task 基类。4.3 注意事项任务函数体 ≠ 任务耗时排队、预取、反序列化、信号、Backend 写入都在耗时里——容量公式第 30 章用「端到端」而非函数体时长。改源码调试后必须还原git diff 检查生产不带实验代码。ExceptionInfo 的序列化依赖异常对象可 pickle自定义异常要兼容。慢任务阈值按 P99 校准别拍脑袋定 3s。trace 是「只读理解」的禁区业务侧通过信号第 26 章介入别在任务代码里改 trace 行为。4.4 常见踩坑经验3 个生产故障故障报表任务「执行 1 秒、总耗时 10 分钟」。根因任务在队列里排队 9 分 59 秒预取被长任务占满第 18 章。对策队列隔离 预取调 1。教训耗时排查先分「排队段」与「执行段」第 15 章三板斧。故障慢任务告警天天响值班脱敏。根因阈值 3s 低于 P99。对策用历史数据定 P99阈值 P99 × 1.5。教训告警阈值脱离数据 狼来了。故障get() 抛出的异常和 Worker 里的对不上。根因自定义异常类不可序列化ExceptionInfo 降级为通用错误。对策异常类保持简单可 pickle。教训跨进程的异常也是「数据」要可序列化。故障任务里发子任务子任务全堆在默认队列没人消费。根因父任务在 order 队列子任务没指定队列第 6 章调用选项没传。对策子任务显式带 queue 或路由按任务名覆盖。教训「任务里发任务」的队列归属要显式声明别指望继承父队列。4.5 思考题build_tracer是「按任务生成执行器」的工厂——同一个任务被两个 Worker 同时执行tracer 会生成几次提示缓存与线程局部trace_task里「写 Backend」失败Redis 挂了会发生什么任务会判失败吗还是结果丢了任务照样成功答案见第 35 章开头的「上一章思考题参考答案」。执行引擎看完了下一站是「消息在网上长什么样」——第 35 章报文与序列化。延伸阅读与资源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 实战修炼与源码剖析