MQ、回调与补偿任务地图

MQ 问题不能只问“消息发没发”。完整问题是：业务事务何时提交、消息何时发布、路由到了哪个消费者、消费者怎样定位业务单、失败返回 ACK 还是 NACK、重复投递是否安全、最终状态是否需要补偿。

本文覆盖 DGJ2.0 与支付中心、订单中心、调拨/配送中心、OA、商品中心、SAAS、App、渠道平台、对账单任务之间的异步链路。目标是让开发、测试和运维可以从 destination、routing key、业务单号、消息体任一线索，追到生产者、消费者、表、状态和补偿入口。

1. 业务目标

• 建立 destination -> routing key -> producer -> consumer -> callback -> 业务表 的完整地图。
• 写清主要消息 payload 的字段、关联键和单位，而不是只列事件名称。
• 区分 Direct、Topic、延迟队列、外部回调和普通 CLI 任务。
• 解释 ACK、NACK、异常被吞掉、主动忽略、迁移后空消费的差异。
• 明确消息 ID、业务幂等键、状态前置条件、数据库唯一键各自解决什么问题。
• 给出“未消费、重复消费、部分成功、乱序、迟到、毒消息”的排查和恢复步骤。
• 补偿前先证明可重跑，补偿后核对主表、明细、库存、资金和下游状态。

2. 边界与基本概念

2.1 本文包含

• application/KzData/Enums/MqEventEnums.php 中的主要 destination 和 routing key。
• application/Services/Mq/MqSer.php 中的主要生产者。
• application/controllers/tasks/*Notify.php 中的主要消费者。
• 秒杀、对账单、库存预警等补偿/异步任务。
• 消息回调事务、幂等、重试、延迟、人工重放和验证。

2.2 本文不直接给出

• RabbitMQ 账号、密码、主机等敏感配置值。
• 生产队列积压量、消费者实例数和真实死信策略；这些必须在目标环境确认。
• 外部系统内部的消费实现，只描述 DGJ 已知的请求/响应边界。
• 直接可执行的生产数据修复 SQL；本文只给只读检查和安全门槛。

2.3 四个不能混用的标识

3. 消息生命周期总图

flowchart LR
    A["业务事务"] --> B["构造 payload"]
    B --> C["MqSer / 外部生产者"]
    C --> D["Exchange"]
    D --> E["destination.routing-key"]
    E --> F["Queue"]
    F --> G["tasks Consumer"]
    G --> H["registryCallback"]
    H --> I["业务状态/明细/库存/资金"]
    I --> J{"处理结果"}
    J -->|ACK| K["消息确认"]
    J -->|NACK| L["重投/死信，依环境配置"]
    J -->|异常被捕获后 ACK| M["消息丢弃，靠告警或补偿"]

3.1 正常消费的七个证据

1. 生产业务事实存在。
2. 生产者方法确实执行到 publish()。
3. destination 和 routing key 与消费者绑定一致。
4. 消费进程正在运行。
5. 回调日志能找到业务键。
6. 数据库事务提交，业务状态和副作用一致。
7. 回调返回 ACK，或重复消息按设计被安全 ACK。

只有第 7 项没有前六项，不能证明业务完成；只有业务表变化但没有 ACK，也可能在稍后重复消费。

4. destination 总表

5. 生产者模型

5.1 核心代码

5.2 Direct 与 Topic

业务代码使用哪种 Producer 不足以证明 RabbitMQ 真实绑定；exchange、queue 和 binding 必须在环境中核验。

5.3 消息 ID 风险

MqSer 的大多数方法把 uniqid() 传给 Producer：

createTopicProducer(destination, routingKey, uniqid())
createDirectProducer(destination, routingKey, uniqid())

这意味着：

• 同一业务操作重发两次会得到两个不同消息 ID。
• 消费者不能仅按 message ID 判断业务重复。
• 需要以订单号、来源单号、明细行号、任务 ID、状态或数据库唯一键做幂等。
• 人工重放时更不能假设旧 message ID 会阻止重复副作用。

5.4 发布与事务的三种边界

flowchart TD
    A["业务写库"] --> B{"MQ 在何时发送"}
    B -->|事务提交后| C["库成功、发送失败 -> 需要补偿"]
    B -->|事务提交前| D["消息可能先被消费 -> 读不到已提交数据"]
    B -->|事务内部同步发送| E["MQ 超时拉长事务和锁"]
    C --> F["Outbox/任务表/状态扫描更可靠"]
    D --> G["消费者重试或延迟"]
    E --> H["确认回滚后消息是否已经发出"]

本文只能从调用位置静态判断单个链路。任何重要改动都应在具体 Service 中确认事务开始、提交和 publish() 的先后关系。

6. 支付中心消息

6.1 消费入口

destination: paycenter_notify
consumer: application/controllers/tasks/PayCenterNotify.php
start: php index.php tasks/PayCenterNotify/consume

6.2 routing key 与回调

6.3 支付成功 payload 骨架

{
  "sourceOrderNo": "<采购单或预订单号>",
  "sourceOrderType": "<01采购单/04预订单>",
  "payOrderNo": "<支付中心支付单号>",
  "mainPayTypeCode": "<主支付方式>",
  "payStatus": "02",
  "payStatusMsg": "支付成功",
  "payTime": "YYYY-MM-DD HH:mm:ss",
  "detailTotalAmount": "<支付中心金额口径>",
  "details": [
    {
      "payAccountTypeCode": "<账户类型>",
      "payAccountTypeDesc": "<账户说明>",
      "payAmount": "<分账户金额>"
    }
  ]
}

必须与支付中心合同确认金额单位。DGJ 内部不同表可能使用元、分或历史字符串金额，不能只凭字段名换算。

6.4 支付回调数据流

sequenceDiagram
    participant P as 支付中心
    participant Q as paycenter_notify
    participant C as PayCenterNotify
    participant O as 采购/预订单
    participant F as Payment/PaymentInfo
    participant R as TaskRecord
    P->>Q: paycenter_pay_result
    Q->>C: payload
    C->>C: payStatus 必须为 02
    C->>O: 按 sourceOrderNo/sourceOrderType 找单
    C->>O: 检查当前状态与支付方式
    C->>F: 写支付主表、明细、采购支付关系
    C->>O: 推进订单状态/额度/活动订单
    C->>C: 提交事务
    C->>R: 查询是否有后置任务记录
    C-->>Q: ACK 或 NACK

6.5 支付幂等和危险分支

6.6 核心表

7. 订单中心消息

7.1 消费入口

destination: ordercenter_notify
consumer: application/controllers/tasks/OrderCenterNotify.php

7.2 注册事件

7.3 所有权迁移风险

以下四个回调当前在方法开头直接 ACK，后面的 DGJ 历史代码不可达，并注明已迁移到采购服务：

• orderCenterAfterSaleAudit
• orderCenterAfterSaleClose
• orderCenterAfterSaleUpdate
• orderCenterAfterSaleConfirmWareHouse

flowchart LR
    A["订单中心售后消息"] --> B["DGJ OrderCenterNotify"]
    B --> C["立即 ACK"]
    C --> D["DGJ 历史处理代码不可达"]
    A --> E["dgj-purchase-service 应承担实际消费"]

排查这四类事件时：

1. DGJ 返回 ACK 不能证明售后已处理。
2. 必须查采购服务的队列绑定、消费日志和数据库。
3. 若两个服务都绑定同一 Topic，要确认是各自独立队列还是竞争消费。
4. 不要通过删除 DGJ 的早返回来“恢复”历史逻辑，除非完成跨服务所有权设计。

7.4 订单中心主数据关系

flowchart TD
    A["订单中心订单号/售后号"] --> B["outOrderNo/outAftersaleNo"]
    B --> C["DGJ SCM_PO_ORDER.billNo"]
    C --> D["SCM_PO_ORDER_INFO"]
    D --> E["sourceLineCode / srcOrderEntryId"]
    E --> F["关闭明细/数量表/库存预警事件"]

消息体中的 sourceChannel、外部售后号、行号、审核结果和关闭明细必须与 DGJ 订单来源、交易类型和当前状态共同校验。

7.5 事务和后置 MQ

历史回调中常见模式是：

1. 开启采购 Model 事务。
2. 更新采购主单、明细和关闭记录。
3. 提交事务。
4. 发送库存事件，重算库存预警。

如果第 4 步失败，采购事实可能已提交但预警未刷新。此时应补发库存事件或重算预警，不应重做整个售后业务。

8. 调拨/配送中心消息

8.1 入口和事件

8.2 出库链路

sequenceDiagram
    participant D as 调拨/配送中心
    participant C as DispatchCenterNotify
    participant P as 采购主单/明细
    participant O as 采购出库记录
    participant I as 库存/后置同步
    D->>C: stock_out payload
    C->>P: 按单号、来源、状态找采购单
    C->>O: 写出库数量和出库记录
    C->>P: 更新待出库/配送中/完成
    C->>C: 提交事务
    C->>I: 库存、消息、活动等副作用
    C-->>D: ACK/NACK

8.3 重复与迟到消息

• 已完成的出库事实再次到达时，应通过已出库数量和状态避免重复累加。
• 订单已取消或关闭后收到出库消息，需要明确谁拥有最终事实，不能无条件恢复订单。
• 物流通知可能先于业务出库落库，需要消费者允许重试或暂存。
• 部分分支对“目标不存在/已处理”直接 ACK，这是主动终止重试，不等于处理成功。

9. OA 消息

9.1 入口

destination: oa_notify
routing key: oa_audit
consumer: tasks/OaNotify::oaAudit

OaNotify 根据模板/回调 code 分发到不同业务。秒杀活动使用 FlashActivityEnums::OA_TEMPLATE_CODE，然后进入 FlashActivitySer::callback20260526。

9.2 秒杀 OA payload 骨架

不同 OA 平台版本字段可能不同，DGJ 至少需要识别：

{
  "callback": "<模板或业务回调code>",
  "status": "<审批结果>",
  "form_data": {
    "activity_id": "<秒杀活动ID>"
  },
  "operator": "<审批人或系统>"
}

实际字段必须以生产消息和 oaAudit 分支为准，不应根据此骨架直接伪造回调。

9.3 OA 状态数据流

flowchart LR
    A["OA 审批结果"] --> B["OaNotify::oaAudit"]
    B --> C{"callback/template"}
    C -->|秒杀| D["FlashActivitySer::callback20260526"]
    D --> E["FLASH_ACTIVITY 状态"]
    D --> F["FLASH_ACTIVITY_LOG"]
    C -->|其他业务| G["对应 Service 回调"]
    B --> H{"异常"}
    H -->|抛出| I["NACK"]
    H -->|正常| J["ACK"]

重复 OA 回调必须满足：已通过再次通过不重复启用、不重复写数量；驳回后迟到通过不能绕过当前状态；活动已结束后审批结果不能把活动恢复运行。

10. 商品中心消息

10.1 事件地图

10.2 数据传播

flowchart TD
    A["商品中心事件"] --> B["ItemCenterNotify"]
    B --> C["商品/套包/限购/价格表"]
    C --> D["商品缓存"]
    C --> E["ES/搜索数据"]
    C --> F["采购可买校验"]
    C --> G["销售可卖校验"]
    C --> H["活动商品校验"]

10.3 ACK 策略风险

ItemCenterNotify 很多校验失败、数据不存在和异常分支仍返回 ACK。这样做可以避免毒消息无限重试，但代价是：

• 消息不会自动再次投递。
• 失败必须依赖日志、告警或定时全量同步发现。
• 重发前要确认依赖数据已经准备好。
• 消费成功不能只看队列清空，还要查缓存/索引和最终业务效果。

套包同步使用数据库事务，但异常回滚后也可返回 ACK。该链路应配套失败记录或重建任务。

11. DGJ 内部 dgj 消息

11.1 消费入口

SaasOrderNotify 消费 DEST_DGJ 下多类事件：

11.2 库存事件 payload

生产者 sendInventoryEvent 明确构造：

{
  "inventoryData": [
    {
      "invId": "<库存商品ID>",
      "qty": "<变动数量及其他上下文字段>"
    }
  ],
  "iid": "<库存流水或删除撤销关联ID，未使用为0>",
  "billType": "<单据类型>",
  "transType": "<交易类型>",
  "sid": "<服务站ID>",
  "po_order_id": "<采购单ID，未使用为0>"
}

11.3 库存预警重算公式

inventory_event 不直接重做库存流水，而是收集受影响 invId，读取实时库存和采购在途后重算预警：

warning_num = 实时库存 qty + 在途 waitQty - 预警下限 lowQty
suggest_num = 预警上限 highQty - 实时库存 qty - 在途 waitQty

flowchart TD
    A["inventoryData / iid / po_order_id"] --> B["汇总 invId"]
    B --> C["商品状态和预警上下限"]
    C --> D["SCM_INVENTORY_REAL_TIME"]
    C --> E["采购在途数量"]
    D --> F["warning_num / suggest_num"]
    E --> F
    F --> G["SCM_INVENTORY_WARNING batchUpsert"]

空 inventoryData、iid=0、po_order_id=0 时直接 ACK；找不到有效商品也 ACK。排查预警不更新时，应确认 payload 至少携带一种可定位方式。

11.4 OPS 批处理风险

opsCreateOrder 对 ids 循环，单个 ID 失败只记录日志后继续，最终整体 ACK；opsCloseOnlineOrder 也会把失败状态落到日志表后整体 ACK。

这属于“部分成功 + 不自动重试”模型：

• 不能根据 ACK 判断全部 ID 成功。
• 必须查每个 ID 的处理状态。
• 补偿应只重跑失败 ID，不能重发整批。

12. DGJ 发往 App/SAAS 的消息

12.1 主要生产者

新销售同步还会使用 SyncOrderBuilder 构造维修厂销售单/出库单事件。排查下游未更新时，要区分旧 APP_SA_* routing key 和新 garage_repair_* 事件。

12.2 下游同步链路

flowchart LR
    A["销售单/出库单事务"] --> B["SyncOrderBuilder"]
    B --> C["统一来源单号、明细、状态"]
    C --> D["dgj_notify"]
    D --> E["App/SAAS 消费者"]
    E --> F["下游订单事实"]
    F --> G["下游后续重试/回执"]

核对字段至少包括：DGJ 订单号、出库单号、来源单号、服务站、客户、明细行、SKU、数量、状态、事件时间。只按订单号查不到时，应继续按来源单号和出库单号追踪。

13. 渠道消息

13.1 事件与处理

13.2 ACK 语义

渠道消费者会对某些“记录不存在、状态已处理、不需要动作”情况直接 ACK。这是业务性忽略，目的通常是阻止无意义重试。

flowchart TD
    A["渠道消息"] --> B{"关系单存在?"}
    B -->|否| C["记录日志并 ACK"]
    B -->|是| D{"状态允许?"}
    D -->|已完成/无需处理| E["ACK"]
    D -->|允许| F["调用渠道 Service"]
    F -->|成功| G["ACK"]
    F -->|可重试异常| H["部分方法 NACK"]

若业务方认为“记录不存在”应该重试，就不能只改 ACK 为 NACK，还要解决消息早到、关系表创建时序和无限毒消息问题。

14. 对账单异步任务

14.1 三条队列

14.2 任务化数据流

flowchart LR
    A["对账策略"] --> B["任务初始化"]
    B --> C["StatementBillTask"]
    C --> D["statement_bill_create"]
    D --> E["生成主单/明细/明细项"]
    E --> F["核对/生成PDF"]
    F --> G["statement_send"]
    G --> H["发送记录/机器人/微信"]
    E --> I["statement_bill_receive_status"]
    I --> J["收款状态"]

StatementTaskSer 和发送记录表是天然补偿抓手。补偿应按任务 ID 或对账单 ID 执行，不能仅按日期整批重发。

14.3 部分成功

对账单批量发送可能出现：部分对账单不存在、未核对、机器人离线、客户未绑定微信、指定绑定无效。正确结果应包含成功、跳过和失败明细，而不是把整批简化为一个布尔值。

15. 延迟队列

15.1 已确认的 TTL + DLX 模型

sequenceDiagram
    participant P as 生产者
    participant D as 延迟队列
    participant X as Direct Exchange
    participant C as 正式消费者
    P->>D: delay routing key
    Note over D: x-message-ttl
    D->>X: dead-letter routing key
    X->>C: 正式事件
    C-->>X: ACK/NACK

15.2 延迟队列注意事项

• 队列声明发生在消费者初始化代码，进程从未启动时队列可能不存在。
• 代码中的 TTL 不代表线上队列参数一定一致，应读取实际队列声明。
• 延迟消息到期时业务状态可能已被人工操作改变，消费者必须再次检查状态。
• 重复发送延迟消息会产生多次到期事件，业务层必须幂等。

16. ACK、NACK 和异常语义

16.1 五种结果

16.2 ACK 不能证明什么

• 不能证明每个批次子项成功。
• 不能证明后置 MQ 成功。
• 不能证明下游系统已落库。
• 不能证明消费者没有走“迁移后直接 ACK”。
• 不能证明缓存、索引、报表已经更新。

16.3 NACK 前必须考虑

flowchart TD
    A["消费者异常"] --> B{"重试能改变结果吗?"}
    B -->|依赖暂时不可用| C["NACK 合理"]
    B -->|永久脏数据| D["无限 NACK 形成毒消息"]
    B -->|业务已部分成功| E["直接重试可能重复副作用"]
    D --> F["死信/失败记录/人工修复"]
    E --> G["先识别已完成步骤，再补剩余步骤"]
    C --> H["确认退避、次数和告警"]

17. 幂等设计字典

17.1 幂等层级

17.2 推荐幂等键

18. 常见故障树

18.1 消息没有到消费者

flowchart TD
    A["没有消费日志"] --> B{"生产日志有 publish 吗?"}
    B -->|否| C["业务分支未发送/事务异常"]
    B -->|是| D{"destination/routing key 正确?"}
    D -->|否| E["常量或调用参数错误"]
    D -->|是| F{"队列绑定存在?"}
    F -->|否| G["部署/声明/环境配置"]
    F -->|是| H{"消费者进程运行?"}
    H -->|否| I["启动/进程守护/崩溃"]
    H -->|是| J["积压、未注册回调、日志路径"]

18.2 消息重复消费

1. 查生产端是否执行了两次 publish()。
2. 查消费者第一次是否 NACK 或进程在 ACK 前退出。
3. 查队列是否有重试/死信回投。
4. 比较业务键，不要只比较随机 message ID。
5. 查第一次事务是否提交。
6. 查副作用是否有唯一引用。
7. 只补未完成步骤。

18.3 队列清空但业务没变化

优先查以下主动 ACK 分支：

• 订单中心售后四个已迁移方法直接 ACK。
• 商品中心异常/数据不存在后 ACK。
• 渠道记录不存在或状态无需处理后 ACK。
• OPS 批次子项失败但整体 ACK。
• 库存事件没有任何定位字段时 ACK。

18.4 部分成功

flowchart LR
    A["消息包含 N 个子项"] --> B["逐项处理"]
    B --> C["成功集合"]
    B --> D["失败集合"]
    C --> E["已产生业务副作用"]
    D --> F["日志/失败状态"]
    F --> G["按失败 ID 精确补偿"]
    G --> H["不得整批盲重放"]

19. 排查 SOP

19.1 收集事实

先记录：

环境：
发现时间：
destination：
routing key：
业务单号/来源单号：
sid：
消息 ID（如有）：
请求或事件时间：
预期状态：
实际状态：
是否有库存/资金副作用：

19.2 从代码确认路由

rg -n "DEST_PAYCENTER_NOTIFY|TYPE_PAYCENTER_PAY_RESULT" application/KzData/Enums/MqEventEnums.php
rg -n "createTopicConsumer|registryCallback" application/controllers/tasks/PayCenterNotify.php
rg -n "createDirectProducer|createTopicProducer" application/Services/Mq/MqSer.php

19.3 检查消费者进程

ps aux | rg 'index.php tasks/.+Notify/consume|Statement.*Consumer'

具体进程管理器名称、容器和日志路径以部署手册为准。

19.4 查日志

组合条件：

消费者类名 + 回调方法
routing key + 业务单号
sourceOrderNo + payOrderNo
sid + 采购单号/销售单号
消息到达时间前后 5-10 分钟

不要只搜“error”；很多消费者使用 info/warn 输出失败，部分还使用标准输出。

19.5 查数据库事实

以下 SQL 只读，分表后缀和字段以目标环境表结构为准：

-- 采购主单
SELECT id, sid, billNo, billStatus, orderType, transType, srcOrderId, srcOrderType, updateTime
FROM t_scm_po_order_<shard>
WHERE billNo = '<业务单号>';

-- 采购支付关系
SELECT *
FROM t_scm_po_order_payinfo_<shard>
WHERE po_order_id = <采购单ID>;

-- 支付主单和明细
SELECT * FROM t_scm_payment_<shard> WHERE srcOrderNo = '<业务单号>';
SELECT * FROM t_scm_payment_info_<shard> WHERE payment_id = <支付主单ID>;

-- 秒杀订单
SELECT id, order_no, po_order_id, status, locked_qty, used_qty, update_time
FROM t_flash_order
WHERE order_no = '<秒杀单号>' OR po_order_id = <采购单ID>;

-- 库存预警
SELECT sid, inv_id, sku_id, warning_num, suggest_num
FROM t_scm_inventory_warning
WHERE sid = <sid> AND inv_id IN (<invIds>);

19.6 判断恢复方式

20. 补偿任务地图

20.1 秒杀

20.2 消费者内置 resend

部分消费者包含 resend()，例如支付、订单中心、商品中心、配送和 SaasOrderNotify。这类方法通常从 /tmp 文件读取消息并重新发布，属于人工运维工具，不是自动补偿保证。

执行前必须：

1. 阅读具体 resend() 的允许 routing key。
2. 确认输入文件格式和每行消息体。
3. 去除已成功业务键。
4. 在非生产环境验证一条。
5. 记录原始消息、执行人、时间和数量。
6. 执行后逐条核对业务结果。
7. 删除或归档临时文件，避免误重跑。

20.3 补偿安全门

flowchart TD
    A["准备补偿"] --> B["冻结原始证据"]
    B --> C["导出业务键和当前状态"]
    C --> D["证明操作幂等"]
    D --> E["先单条/小批量"]
    E --> F["核对主表、明细、副作用"]
    F --> G{"结果正确?"}
    G -->|否| H["立即停止，恢复/人工分析"]
    G -->|是| I["扩大批次"]
    I --> J["最终对账和执行记录"]

21. 高风险清单

22. 回归清单

22.1 生产者

• [ ] 正常业务只发布一次预期事件。
• [ ] 事务回滚时不会留下不可撤销消息。
• [ ] 事务提交后发布失败有补偿抓手。
• [ ] payload 包含稳定业务键、sid、来源和事件时间。
• [ ] 新字段对旧消费者可选兼容。
• [ ] 不在 payload 和日志中发送密钥或隐私数据。

22.2 消费者

• [ ] 正常消息推进主表、明细和副作用。
• [ ] 相同业务消息重复投递不会重复记账、扣库存或建单。
• [ ] 乱序消息不会把终态回退到中间态。
• [ ] 业务单暂时不存在时重试策略明确。
• [ ] 永久脏消息不会无限 NACK。
• [ ] ACK、NACK 和主动忽略都有日志和指标。
• [ ] 批处理能记录每个子项结果。
• [ ] 事务提交与 ACK 先后关系经过故障注入验证。

22.3 延迟和补偿

• [ ] 延迟队列 TTL、DLX 和正式 routing key 与代码一致。
• [ ] 同一延迟事件重复到期仍幂等。
• [ ] 补偿任务 limit 有上限并可小批运行。
• [ ] 补偿失败能被调度系统识别，不只看退出码。
• [ ] 人工重放只包含失败业务键。
• [ ] 补偿后完成业务、库存、资金和下游四层对账。

22.4 关键专项

• [ ] 支付成功、失败、异常、重复成功、迟到成功、退款分别验证。
• [ ] 订单中心审核、关闭、售后所有权迁移分别验证。
• [ ] 商品限购、套包、价格生效/失效、敏感词分别验证。
• [ ] 配送出库、物流早到、自提、退货关闭分别验证。
• [ ] 秒杀超时与支付回调并发验证。
• [ ] 对账单生成、部分发送失败、重复发送分别验证。

23. 已确认与待环境确认

23.1 静态代码已确认

• MqEventEnums 集中定义主要 destination 和 routing key。
• MqSer 同时使用 Direct 和 Topic Producer。
• 大多数生产者使用随机 uniqid() 作为消息 ID。
• PayCenterNotify 注册 10 类支付中心事件。
• 订单中心四个售后回调已迁移并在 DGJ 直接 ACK。
• ItemCenterNotify 多个失败分支返回 ACK。
• SaasOrderNotify::inventory_event 重算预警而非重做库存流水。
• 新站 300 秒和暂停取消 60 秒延迟队列使用 TTL + dead-letter routing。
• 对账单拆为策略、生成/收款状态、发送三类 destination。
• 多个消费者提供基于临时文件的人工 resend()。

23.2 必须在环境确认

• RabbitMQ exchange、queue、binding、durable、prefetch、消费者并发配置。
• NACK 是重新入队、进入重试队列还是死信。
• 自动重试次数、退避策略和死信告警。
• 各消费者进程管理器、实例数、日志路径和告警规则。
• dispatch_out_delay_20 的实际 TTL 和 DLX。
• 消息生产是否启用 publisher confirm。
• 消费 ACK 是否在 callback 返回后可靠执行。
• 采购服务对已迁移订单中心事件的真实队列绑定。
• 核心业务表是否有业务唯一键或事件处理记录。
• 对账、商品中心和渠道 ACK 丢弃后的定时兜底任务。

24. 证据来源

25. 一页式排查结论

先用 destination + routing key 找消费者，不要先猜业务表。
没有消费日志：查 publish、binding、进程和注册回调。
有消费日志：查业务键、事务提交和 ACK/NACK。
队列清空但业务没变：重点查主动 ACK、迁移空消费和部分成功。
重复消费：比较业务键，不比较随机 message ID。
主业务完成但派生数据缺失：只补后置 MQ/缓存/预警，不重做主单。
任何人工重放：先导出当前状态，只重放失败键，小批验证后扩大。

请求-日志-数据变更追踪卡

多入口请求链路

日志证据矩阵

| 链路段 | 日志来源 | 可检索锚点 | 成功信号 | 失败信号 | 与下一段关联方式 | | --- | --- | --- | --- | --- | --- | --- | | 生产 | MqSer/Provider 日志 | destination、routing key、业务单号、message ID | Broker 接受/生产成功 | 序列化失败、连接错误、发送异常 | message ID + 业务键查消费端 | | 消费入口 | *Notify 方法 | destination、event、消息 ID、来源单号 | 命中目标 handler | 未知事件、迁移空消费 | handler 名 + 业务键进入 Service | | 业务处理 | Consumer 调用的领域 Service | 业务单号、当前状态、幂等键 | commit 且状态/数量正确 | rollback、找不到关系、顺序冲突 | 单号回查主表、流水、关系表 | | ACK/NACK | Consumer 框架/Broker | message ID、重试次数 | ACK；重复消费零副作用 | NACK、反复重投、死信 | 重试记录与同一业务键对照 | | 补偿 | Task 日志 | task、批次、业务 ID、扫描/成功/失败数 | 重跑后失败数下降，已完成跳过 | 扫描无边界、重复扣加、任务中断 | 批次清单回查业务和消息事实 |

环节数据变更台账

子模块追踪：paycenter-mq 支付中心消息

子模块追踪：ordercenter-mq 订单中心消息

子模块追踪：dispatch-mq 调拨与配送中心消息

子模块追踪：oa-mq OA 消息

子模块追踪：itemcenter-mq 商品中心消息

子模块追踪：dgj-mq DGJ 内部消息

子模块追踪：saas-mq DGJ 发往 App 与 SAAS 的消息

子模块追踪：channel-mq 渠道消息

子模块追踪：statement-mq 对账单异步任务

子模块追踪：delay-compensate 延迟队列、重试与补偿任务