Python后端中间件专题15一条坏消息堵住整条队列——有限重试、DLQ 与重放值班人员收到一条“通知队列持续失败”告警。如果 Worker 没有终点一条缺字段的 JSON 会被反复投递占用执行槽并刷屏日志。如果直接 ACK 并丢弃现场证据和可恢复性又消失。DLQ 的价值不是“另一个垃圾队列”而是给失败工作一个有限、可审计、可操作的去处。值班 runbook先隔离再定点重放按reason、schema version、producer 与时间窗口聚合 SQLdead_letters阻断仍在制造毒消息的来源保存原 body 和 error。对指定 dead-letter ID 核对修复条件不批量重放异质消息。条件 claim 必须有 lease expiry防两个操作员并发获得同一行。保留原 event ID 和 body附加重放来源 header以 confirmed publisher 发回主队列确认后才写replayed_at。发布失败释放 claim确认不确定仍可能重复由消费 receipt 收敛。再次操作同一 ID 应返回not_replayed并核对最终业务 effect 只有一份。若未运行真实 broker/SQL 集成这四步只是操作合同不是完成记录。runbook 的学习目标你将能按 reason 分类 poison/permanent/exhausted 死信保留原始 event body 并重置 delivery 计数设计“按死信 ID 修复后重放”的运维流程并说明重放本身为什么也要 confirm、claim 和幂等。值班前置与授权边界请了解 12 的有界重试和 14 的 confirm 不确定窗口。manifest selector 需远程 RabbitMQ、Worker、PostgreSQL 和真实任务提交本次禁止远程与 Docker因此不会运行它。上一课练习答案答案 EX-14-01 [code]在project/下保存dlq_replay_probe.pyfromticketflow.messaging.dead_letterimportDeadLetterMessage,replay_message originalDeadLetterMessage(bodyb{event_id:event-15},reasonretries_exhausted,attempts3,headers{x-tenant:tenant-a},)replayedreplay_message(original)assertreplayed.bodyoriginal.bodyassertreplayed.attempts0assertreplayed.headers[x-tenant]tenant-aassertreplayed.headers[x-replayed-from-dlq]trueassertreplayed.headers[x-dead-letter-reason]retries_exhaustedprint(replayed.body.decode())print(replayed.attempts,replayed.headers)运行命令$env:PYTHONPATHsrc; python dlq_replay_probe.py预期输出{event_id:event-15} 0 {x-tenant: tenant-a, x-replayed-from-dlq: true, x-dead-letter-reason: retries_exhausted}重放不改写 event ID 或 body因为它仍是同一个业务事件。它只重置本次 delivery 的计数并附加运维来源使后续日志能区分首次投递与人工重放。答案 EX-14-02 [prose]poison_message先保留原 JSON 和 parse error核对 producer schema/version只有在能恢复为完整 envelope 后才重放。permanent_failure先判断业务规则、租户边界或不支持 event type 是否真已修复如果事件本身不再合法应终结并审计不强行重放。retries_exhausted先确认依赖已恢复并有容量再分批重放否则会立即再次压垮依赖。不能全量一键重放因为 DLQ 里的 reason、schema 和故障时间可能不同旧毒消息会再次占满队列而突发重放还可能造成下游流量洪峰。为什么隔离不能直接画成“已修复”Worker 在 parse 失败、永久错误或 retry 耗尽后把 original primitive event、reason、attempts 和 error 持久化到 SQLdead_letters。RabbitMQ 的 dead queue 可承载一个可观测指针但 SQL row 才是本项目的可查询重放事实。这样即使发布 DLQ pointer 失败操作员仍能检索和修复原始记录。重放不应直接把 row 标记为 replayed 后再 publish。正确顺序是先用有过期时间的 claim 防止并发操作员重复发布通过 confirmed publisher 发回主队列成功后才设replayed_at。发布失败则释放 claimrow 仍可重试。即使两个操作员在极端 confirm 不确定窗口中仍产生重复原 event ID 未改变13 的 receipt 会将业务副作用收敛为一份。runbook 的证据分层本地可运行的 primitive replay 合同python -m pytest tests/unit/test_messaging_core.py::test_dlq_replay_preserves_envelope_and_resets_delivery_attempt_metadata -q本地可信输出. [100%] 1 passed本课累积 revision 另用 SQLAlchemy 2 SQLite 文件数据库验证原始 body/headers 持久化、原子条件 claim、租约过期后换主、旧 owner 不得 mark/release、确认后 mark、失败后 release 与二次重放 no-op这只证明本地 SQL 状态机。它不替代真实 RabbitMQ/Worker/PostgreSQL 运行manifest 远程 selectorpython -m pytest tests/integration/test_dead_letter.py::test_poison_notification_reaches_dlq_and_replay_applies_one_effect -q仍需验证 poison 持久化、人工修复、重放只产生一份 effect 且第二次重放返回not_replayed。本次远程检查保持PENDING未运行。runbook 中途失败时DLQ 持续增长按 reason/schema version/producer 聚合先停止新毒消息源而不是立即重放存量。replayed_at已有值但下游没有消息检查是否在 confirm 前就标记完成正确实现只能在 confirmed publish 后更新。两个操作员都说重放成功需要 DB 条件 claim而不是先 SELECT 后 UPDATE同时保留 consumer receipt 作最后防线。交接记录有限重试保护容量DLQ 保留证据按 ID 修复重放恢复业务。重放是一次新的传输尝试因此 confirm、claim 和下游幂等一个都不能少。本课练习练习 EX-15-01 [code]实现一个 in-memorySlaStore始终返回同一个(ticket_id, version)但mark_escalated只对首次条件转移返回 True。在同一now连续扫描两次断言 escalated 为 1 然后 0并打印结果。练习 EX-15-02 [prose]比较 Celery Beat “按 UTC 间隔每分钟触发扫描”与“为每张工单安排一个本地时区 ETA task”在时区/DST、丢调度、重复调度和补跑方面的差异给出 TicketFlow 的选择。下一篇预告16 不再等用户请求或消息到来Beat 会重复发起 SLA 扫描。我们会把“凌晨两点”转换成 UTC 事实时间并用条件 UPDATE 保证两次 tick 只升级一次。完整核心模块带 claim、确认与释放的死信重放DLQ payloads preserve the original JSON body for deliberate replay.from__future__importannotationsfromdataclassesimportdataclass,fieldfromdatetimeimportdatetime,timedelta,timezoneimportjsonfromtypingimportCallable,Dict,Protocolfromuuidimportuuid4fromsqlalchemyimportor_,updatefromsqlalchemy.ormimportSessionfromticketflow.storage.modelsimportDeadLetterModeldataclass(frozenTrue)classDeadLetterMessage:body:bytesreason:strattempts:intheaders:Dict[str,str]field(default_factorydict)dataclass(frozenTrue)classReplayMessage:body:bytesattempts:intheaders:Dict[str,str]defreplay_message(message:DeadLetterMessage)-ReplayMessage:Reset delivery accounting while retaining the immutable original event body.headersdict(message.headers)headers[x-replayed-from-dlq]trueheaders[x-dead-letter-reason]message.reasonreturnReplayMessage(bodymessage.body,attempts0,headersheaders)dataclass(frozenTrue)classClaimedDeadLetter:The durable row a single operator currently owns for replay.id:strmessage:DeadLetterMessage lease_owner:strclassDeadLetterStore(Protocol):defclaim_for_replay(self,*,dead_letter_id:str,lease_owner:str,now:datetime,lease_for:timedelta)-ClaimedDeadLetter|None:Atomically claim an unreplayed or expired-lease row....defmark_replayed(self,*,dead_letter_id:str,lease_owner:str,at:datetime)-bool:Record completion only for the active claim after publisher confirmation....defrelease_replay(self,*,dead_letter_id:str,lease_owner:str,reason:str,now:datetime)-None:Make a failed publication eligible for a later deliberate retry....classConfirmingReplayPublisher(Protocol):defpublish(self,*,body:bytes,headers:Dict[str,str])-bool:Return true only after the primary queue publisher confirmation arrives....classSqlAlchemyDeadLetterStore:Durable original body and atomic owner-checked replay lease.def__init__(self,*,sessions:Callable[[],Session])-None:self._sessionssessionsdefpersist(self,*,dead_letter_id:str,message:DeadLetterMessage)-None:withself._sessions()assession:withsession.begin():session.add(DeadLetterModel(iddead_letter_id,bodymessage.body,reasonmessage.reason,attemptsmessage.attempts,headers_jsonjson.dumps(message.headers,sort_keysTrue),))defclaim_for_replay(self,*,dead_letter_id:str,lease_owner:str,now:datetime,lease_for:timedelta)-ClaimedDeadLetter|None:ifnotlease_ownerorlease_fortimedelta(0):raiseValueError(owner and positive lease are required)withself._sessions()assession:withsession.begin():claimedsession.execute(update(DeadLetterModel).where(DeadLetterModel.iddead_letter_id,DeadLetterModel.replayed_at.is_(None),or_(DeadLetterModel.replay_claim_expires_at.is_(None),DeadLetterModel.replay_claim_expires_atnow),).values(replay_claim_ownerlease_owner,replay_claim_expires_atnowlease_for))ifclaimed.rowcount!1:returnNonerowsession.get(DeadLetterModel,dead_letter_id)messageDeadLetterMessage(row.body,row.reason,row.attempts,json.loads(row.headers_json))returnClaimedDeadLetter(row.id,message,lease_owner)defmark_replayed(self,*,dead_letter_id:str,lease_owner:str,at:datetime)-bool:withself._sessions()assession:withsession.begin():completedsession.execute(update(DeadLetterModel).where(DeadLetterModel.iddead_letter_id,DeadLetterModel.replay_claim_ownerlease_owner,DeadLetterModel.replay_claim_expires_atat,DeadLetterModel.replayed_at.is_(None)).values(replayed_atat,replay_claim_ownerNone,replay_claim_expires_atNone,last_replay_errorNone))returncompleted.rowcount1defrelease_replay(self,*,dead_letter_id:str,lease_owner:str,reason:str,now:datetime)-None:withself._sessions()assession:withsession.begin():session.execute(update(DeadLetterModel).where(DeadLetterModel.iddead_letter_id,DeadLetterModel.replay_claim_ownerlease_owner,DeadLetterModel.replayed_at.is_(None)).values(replay_claim_ownerNone,replay_claim_expires_atNone,last_replay_errorreason))classDeadLetterReplayer:Claim → confirmed publish → mark, with release on every publish failure.def__init__(self,*,store:DeadLetterStore,publisher:ConfirmingReplayPublisher,clock:Callable[[],datetime]lambda:datetime.now(timezone.utc),lease_for:timedeltatimedelta(minutes5))-None:iflease_fortimedelta(0):raiseValueError(replay lease must be positive)self._storestore self._publisherpublisher self._clockclock self._lease_forlease_fordefreplay(self,*,dead_letter_id:str,operator_id:str)-bool:nowself._clock()claim_ownerf{operator_id}:{uuid4()}claimedself._store.claim_for_replay(dead_letter_iddead_letter_id,lease_ownerclaim_owner,nownow,lease_forself._lease_for)ifclaimedisNone:returnFalsemessagereplay_message(claimed.message)try:confirmedself._publisher.publish(bodymessage.body,headersmessage.headers)ifnotconfirmed:raiseRuntimeError(publisher confirmation was not received)exceptExceptionaserror:self._store.release_replay(dead_letter_idclaimed.id,lease_ownerclaim_owner,reasontype(error).__name__,nowself._clock())raisereturnself._store.mark_replayed(dead_letter_idclaimed.id,lease_ownerclaim_owner,atself._clock())