系列导航：上一篇 17 从持久日志构建最小消息队列 ｜ 本篇 18 ｜ 下一篇 19 延迟任务的领取与取消

上一篇把“受理”“落盘”“消费确认”拆成了三条线，用一个自制的单机日志逐一观察。换成成熟的消息中间件之后，这三条线依然存在，只是分别落到了客户端协议里：publisher confirm 说明 broker 接下了消息，消费确认把消息的所有权交给消费者，两者之间还隔着队列、重投和消费者自己的数据库事务。中间件不会替应用决定先提交还是先确认，也不会替应用判断一条消息是不是已经处理过。

本篇用真实的 RabbitMQ 3.12.1 节点和 PostgreSQL 16 检验接入方的契约：生产确认与不可路由消息、提交后再确认与去重表、毒消息的有限次重试和隔离、积压时的有界预取。输入是与第 17 章共用的 3000 条合成订单事件，代码在 examples/18-rabbitmq-consumer/，输出在 evidence/18/output.txt。

实验只有一个 RabbitMQ 节点。它能说明单个 broker 进程重启或被 kill -9 之后，已确认的持久消息还在不在；它不能说明多副本持久性，也不能说明集群分区时的行为。第 17 章的单机日志结论同样不能拿来推导这里的保证，两章的保证分别陈述。

需求与不变量

每条事件有一个全局唯一的 message_id，消费者的副作用是给账户累加金额：在 ledger 表写一条流水，在 balance 表累加余额。余额累加不是幂等操作，同一事件施加两次，余额就错了，这让重复处理在数据上可以直接观察到。

接入方要守住的不变量：

  • 发布端：只有收到 basic.ack 的消息才算交给了 broker；连接断开时，未确认的消息按 message_id 重发，宁可重复，不可遗漏。
  • 消费端：同一 message_id 的副作用最多提交一次；副作用提交之前不确认消息。
  • 毒消息（amount_cents 为 null，数据集中有 evt-000137、evt-001291、evt-002718 三条）最多尝试 3 次，之后进入隔离队列，不阻塞其他消息。
  • 积压时，每个消费者在 broker 侧的未确认消息数不超过 prefetch。

规模假设是教学取值：积压 3000 条，每条处理加 3ms 人为延迟，消费者 1 到 4 个，prefetch 取 10 或 50。这些数字只用来观察预取上限的效果，不对应任何生产吞吐。

拓扑与状态归属

图 18-1：拓扑、状态归属与持久化位置（已实现）

1
2
3
4
5
6
7
8
9
10
11
发布端（测试进程）                RabbitMQ 3.12.1 单节点（/tmp/b2-rabbit/mnesia）                  消费者进程 ×N
┌──────────────────┐  publish   ┌──────────┐ "orders" ┌──────────────────────┐  deliver   ┌───────────────────────┐
│ ConfirmPublisher │──────────▶ │ orders.x │────────▶ │ orders.work (classic) │──────────▶ │ Worker（手动确认）     │
│ outstanding:     │◀────────── │ direct   │          │ durable, DLX=retry.x  │◀────────── │ basicQos(prefetch)    │
│  seqNo→message_id│  ack/return└──────────┘          └──────┬───────────────┘   ack/nack  └──────────┬────────────┘
│ （内存，断线重发）│                 ▲                 nack   │ requeue=false                       │ 同一事务
└──────────────────┘                 │ TTL 过期            ┌──▼──────────────┐                ┌────────▼────────────┐
                                     └───────────────────── │ orders.retry    │                │ PostgreSQL b2_18     │
                                                            │ TTL 400ms        │                │ processed(PK)        │
                         第 3 次失败：confirm 发布后再 ack ─▶ │ orders.parking   │                │ ledger / balance     │
                                                            └─────────────────┘                └─────────────────────┘

持久状态分在两处：消息和队列定义在 RabbitMQ 的数据目录，去重记录和副作用在 PostgreSQL。发布端的未确认表只在内存里，发布进程崩溃后要靠上游重新产生事件，本篇没有覆盖这一段。

发布端：确认、退回与重发

发布端打开 confirm 模式，每条消息带 messageId、deliveryMode=2、mandatory=true，并按 channel 上的发布序号记录未确认消息。RabbitMQ 文档规定了 basic.ack 的时机：持久消息路由到 durable 队列时，确认意味着已经落盘；路由到 quorum 队列时，意味着多数副本已经接受（Publisher Confirms，3.13 文档）。本篇引用的是 3.13 版文档，本地运行版本为 3.12.1，复现时需核对版本差异。同一页还说明，消息存储为减少 fsync 次数会按几百毫秒的间隔批量落盘，持续负载下确认延迟可能达到几百毫秒，因此实现采用异步确认加 256 条窗口，不是每发一条就同步等待。

确认不等于入队。unroutable_mandatory_return 把一条消息发往不存在的路由键：

1
CASE unroutable_mandatory_return          PASS basicAck=true basicReturn=true workDepth=0

客户端先收到 basic.return，随后收到 basic.ack，文档对这一顺序有明确规定。只看 ack 的发布端会把这条消息当作发送成功，而它没有进入任何队列。这里 mandatory 和 ReturnListener 是必需的。

broker_restart_persistent_vs_transient 检验确认与持久化标志的组合：500 条持久消息和 200 条非持久消息都收到了确认，两组消息分别放在两个 durable 队列里，平滑重启后前者 500 条仍在，后者为 0。Queues 文档写得很直接：非持久消息即使存放在 durable 队列里，也会在恢复时被丢弃（Queues，3.13 文档）。收到 confirm 只说明 broker 按消息自身的持久化属性处理完毕，持久化属性要由发布端设置。

更强的检验是在发布途中 kill -9 broker。kill9_broker_confirmed_survive 在开始发布 3000 条约 400ms 后杀掉 beam.smp，随后重新启动节点；发布端检测到连接断开，重连成功后重发全部未确认的消息：

1
2
confirmedBeforeKill=1650 missingAfterRestart=0 publishErrors=1 reconnects=1 resent=101 restartMs=14333
workDepthAfterRepublish=3000 processed=2997 ledgerRows=2997 parked=3 consumerDuplicatesSkipped=0

kill 之前确认过的 1650 条在重启后全部存在。重发了 101 条，重启后队列深度恰好是 3000，消费端也没有遇到重复，说明这 101 条在 kill 时都没有被保留，未确认的消息确实可能丢失，重发是必要的。另一种结果同样可能出现：消息已经落盘，只是确认在途中丢了，重发就会造成重复。Reliability Guide 明确提醒了这一点，要求消费端去重或做幂等（Reliability Guide，3.13 文档）。本次运行没有出现这种重复，因此重复路径要靠下一节的消费端实验来覆盖。

图 18-2：发布确认、断线与按 message_id 重发（已实现并经 kill -9 验证）

sequenceDiagram
    participant P as 发布端
    participant R as RabbitMQ
    P->>R: basic.publish seq=1..n（mandatory, persistent, messageId）
    R-->>P: basic.return（仅不可路由时，先于 ack）
    R-->>P: basic.ack seq≤k（multiple，落盘后）
    Note over P: outstanding 中删除 ≤k，记为 confirmed
    P--xR: 连接断开（broker 被 kill -9）
    Note over P: k 之后的消息结果未知
    loop 每 500ms 重连，直到成功
        P->>R: connect
    end
    P->>R: 按 message_id 重发全部未确认消息
    R-->>P: basic.ack
    Note over P,R: 可能重复：确认丢失但消息已落盘，由消费端去重吸收

消费端：提交之后才确认

消费者使用手动确认。每条消息在同一个 PostgreSQL 事务里先插入 processed(message_id)，再写流水、累加余额，提交成功后才调用 basicAck。去重命中时回滚事务，直接确认：

1
2
3
4
5
6
7
8
9
10
11
12
try (PreparedStatement ps = db.prepareStatement(
        "insert into processed(message_id, worker) values (?, ?) on conflict (message_id) do nothing")) {
    ps.setString(1, id); ps.setString(2, name);
    if (ps.executeUpdate() == 0) {          // 已处理过：不施加副作用，直接确认
        db.rollback(); duplicates.incrementAndGet();
        ack(ch, env);
        return;
    }
}
// 写 ledger、累加 balance（同一事务）
db.commit();
ack(ch, env);                               // 提交之后才确认

两个消费者同时处理同一个 message_id 时，后插入的事务会在唯一索引上等待先插入的事务提交或回滚（PostgreSQL 16 §64.5）；对方提交后，on conflict do nothing 让这次插入影响 0 行，于是走去重分支。去重靠数据库唯一约束完成，应用里不需要加锁。本章没有专门构造这种并发命中，只在多消费者用例中观察到副作用没有重复。

确认是有所有权含义的：Reliability Guide 说确认代表“接收以及所有权转移”，消费者在完成写库等操作之前不应确认。手动确认模式下，消费者的 channel 或连接一关闭，所有未确认的投递都会重新入队，再次投递时带上 redeliver 标志（Publisher Confirms，3.13 文档）。crash_after_commit_before_ack 把崩溃放在这个窗口里：消费者在第 10 次提交之后、确认之前 Runtime.halt，prefetch 为 5。

1
2
3
HALT after commit, before ack message_id=evt-000010 requeueObservedMs=2 readyAfterCrash=41
DUPLICATE message_id=evt-000010 redelivered=true
ledgerRows=50 distinct=50 rowsFor_evt-000010=1 redeliveredRows=4 balanceSum=1195882 expectedSum=1195882

崩溃后队列里有 41 条就绪消息：50 条中已确认 9 条，evt-000010 已提交未确认，同它一起被预取但尚未处理的 4 条也回到了队列。第二个消费者收到 5 条带 redelivered=true 的投递，其中 evt-000010 被去重表拦下，其余 4 条是第一次真正处理，因此 redeliveredRows=4。余额与输入的期望值一致。

由此可以看到，redelivered 标志不能用来判断是否重复：带标志的 5 条消息里只有 1 条处理过。文档的措辞是，未设置该标志时可以保证消息没有被投递过，设置了只是一个提示。

去掉去重表的负向对照 negative_no_dedup_double_apply，同样的崩溃得到 51 行流水，evt-000010 出现两次，余额比期望多出 36561，正好等于这条事件的金额。

两种确认顺序也做了对照。先确认后处理（negative_ack_before_process）：第 10 条确认之后、提交之前崩溃，最终只有 49 行，evt-000010 不在数据库里，队列也空了，broker 已经把它当作处理完毕删除。自动确认（negative_autoack_kill9_loses）的结果更严重：

1
published=400 committedAtKill=53 readyBeforeKill=0 readyAfterKill=0 finalLedgerRows=53 lost=347

kill 之前，broker 已经把 400 条全部推给了消费者（readyBeforeKill=0）。自动确认模式下没有预取上限，消息一发出就算投递成功，于是进程只处理了 53 条，剩下 347 条随进程内存一起消失。文档把自动确认称为“fire-and-forget”，认为它不安全，也不能限制消费者的未确认数量。

方案 崩溃窗口 结果（本章观察） 语义
先处理、提交后确认 + 去重表 提交后、确认前 重投 1 次，被去重拦下，副作用 1 次 至少一次投递，副作用一次
先处理、提交后确认，无去重 同上 副作用 2 次 至少一次
先确认、后处理 确认后、提交前 消息丢失 至多一次
自动确认 任意时刻 400 条丢 347 条 至多一次，且无预取上限

去重表与业务天然幂等

Reliability Guide 建议把消费者设计成幂等，而不是显式去重。两种做法各有适用范围。业务天然幂等，指副作用本身可以重复执行，例如“把订单状态置为已支付”，或者带版本条件的更新；它不需要额外的表，但要求每种副作用都能写成幂等形式，余额累加、发送通知这类操作做不到。去重表适用于任意副作用，代价是多一张随消息量增长的表，需要按保留期清理；而且去重键必须和副作用在同一个事务里提交，否则崩溃窗口只是换了个位置。副作用如果落在外部系统，比如调用一个不支持幂等键的接口，同一事务这个前提就不成立，只能要求对方提供幂等键，或者接受至少一次的后果。

本实现用 message_id 做去重键，同时吸收了两类重复：消费确认丢失导致的重投，以及发布端重发导致的重复消息。第 17 章的偏移去重只能挡住前一类，原因在于发布端重试会得到新的偏移。

毒消息：重试队列、原地重入队与 quorum 投递上限

毒消息每次都会处理失败。若用 basicNack(requeue=true) 放回原队列，它会立即再被投递。requeue_hot_loop_vs_retry_queue 用 20 条消息（其中一条是毒消息）运行 2 秒：

1
requeue=true poisonDeliveriesIn2s=5724 otherMessagesCommitted=19

同一条消息在 2 秒内被投递了 5724 次。其余 19 条仍然完成了，没有被它挡住，但这 2 秒里 CPU 和网络都耗在这一条消息上。文档也指出，所有消费者都 requeue 时会形成代价很高的重投循环。

重试队列用 nack(requeue=false) 把消息死信到 orders.retry.x，路由到 TTL 为 400ms 的 orders.retry，消息过期后再死信回 orders.x，重新进入工作队列。尝试次数从 x-death 头里读取：这个头是按时间由近到远排列的数组，每个元素包含来源队列、原因和次数（Dead Letter Exchanges，3.13 文档）。第 3 次失败时，消费者在开启 confirm 的另一个 channel 上把消息发布到 orders.parking，等到确认之后才 ack 原消息，防止消息在“已确认原消息、隔离队列还没收下”这个空档里丢失。

1
2
3
parked=evt-000137 x-attempts=3
x-death=[{queue=orders.retry,reason=expired,count=2}{queue=orders.work,reason=rejected,count=2}]
retries=2 committedWhilePoisonRetrying=871 retryWindowMs=1136

从第一次重试到进入隔离队列历时约 1.1 秒，其间两个消费者提交了 871 条其他消息。这条毒消息被两个消费者轮流处理过，计数仍然正确，因为次数记在消息头里，而不是某个消费者的内存里。

图 18-3：一次投递的处理状态（前置条件与拒绝路径）

1
2
3
4
5
6
7
8
9
投递到达
  ├─ 载荷无效（amount_cents 为 null）
  │    ├─ attempt = x-death[orders.work/rejected].count + 1 < 3 ──▶ nack(requeue=false) → retry（TTL 400ms）→ work
  │    └─ attempt ≥ 3 ──▶ confirm 发布到 parking ──确认──▶ ack 原消息
  │                                         └─未确认/超时──▶ 抛出，channel 关闭，原消息重新入队
  └─ 载荷有效
       ├─ insert processed 影响 0 行 ──▶ rollback，ack（重复）
       ├─ 数据库错误 ──▶ rollback，nack(requeue=true)
       └─ 写 ledger、累加 balance，commit ──▶ ack

quorum 队列自带投递次数上限。3.13 文档说明，quorum 队列记录失败投递的次数，通过 x-delivery-count 头暴露，超过上限的消息会被丢弃，配置了 DLX 时会被死信（Quorum Queues，3.13 文档）。3.12 和 3.13 默认没有上限，必须通过队列参数或 policy 设置；4.0 起默认值才是 20。quorum_delivery_limit_single_node 声明了 x-delivery-limit=2，让消费者对毒消息一直 requeue：

1
deliveries=3 parkedXDeath=[{queue=orders.qq,reason=delivery_limit,count=1}] qqDepth=0

3 次投递之后，消息以 reason=delivery_limit 进入死信队列，最后一次投递携带 x-delivery-count=2（见 evidence/18/logs/21-worker-qq.log）。与重试队列相比，quorum 的上限不需要额外的队列，也不需要消费者自己数次数，但重试之间没有退避间隔，并且要求队列类型是 quorum。

classic durable 队列和 quorum 队列的差别还不止这一项。quorum 队列基于 Raft 复制，它的 publisher confirm 要等多数副本接受后才发出。本节点的 list_queues 显示 orders.qq 只有一个成员 [b2rabbit@localhost]，“多数”就是这一个节点，所以单节点实验能检验投递上限的行为，却不能检验多副本持久性。

积压与有界预取

basicQos(prefetch) 在默认的 global=false 下限制每个消费者的未确认消息数，0 表示不限（Consumer Prefetch，3.13 文档）。backlog_prefetch_bounded 先在队列里积压 3000 条，再启动消费者，同时用 rabbitmqctl list_channels 采样 broker 侧每个 channel 的 messages_unacknowledged：

消费者数 prefetch 采样数 单 channel 未确认最大值 处理 / 隔离 排空观察（approxDrainMs）
2 10 10 10 2997 / 3 10354ms
4 50 12 50 2997 / 3 8615ms
1 0（不限） 8 2962 2997 / 3 16716ms

prefetch 为 10 和 50 时，采样点上的未确认数都没有超过设定值；不设上限时，一个 channel 一次拿走了 2962 条。这些消息堆在消费者的客户端缓冲里，进程崩溃时会全部重投，进程处理不过来时会撑大内存，而且其他消费者再也分不到它们。

排空时间是测试进程的观察值，包含 JVM 启动和约 2 秒一次的 rabbitmqctl 采样间隔；各消费者日志里从首次投递到最后一次提交的 drainMs，在 2×10 配置下约 6.6–6.8 秒，4×50 下约 3.8–4.3 秒，1×0 下约 13.6 秒。每条消息另有 3ms 人为延迟，这些数字只说明本机上增加消费者能缩短排空时间，不能换算成任何生产吞吐。采样只能证明采样点上没有超限，上限本身由 broker 保证。积压恢复时真正要记录的是积压量、消费者数、prefetch、排空所需时间这四个量，再加上隔离队列的增长；改变其中任何一个，都应该重新测量。

用例与证据

用例 类型 观察值（摘自 evidence/18/output.txt）
normal_confirm_consume_poison 正常 + 毒消息 1000 条确认，999 条副作用一次，余额一致，毒消息 3 次尝试后隔离
unroutable_mandatory_return 边界 return 先于 ack，队列深度 0
crash_after_commit_before_ack 恢复 重投 redelivered=true，去重拦下，50 行
negative_no_dedup_double_apply 负向对照 51 行，余额多 36561
negative_ack_before_process 负向对照 49 行，消息丢失
negative_autoack_kill9_loses 负向对照 400 条丢 347 条
broker_restart_persistent_vs_transient 恢复 持久 500→500，非持久 200→0
kill9_broker_confirmed_survive 恢复 已确认 1650 条全部存在，重发 101 条，最终 2997 行
backlog_prefetch_bounded 积压 未确认最大值 10 / 50 / 2962
requeue_hot_loop_vs_retry_queue 方案对比 2 秒内 5724 次投递
quorum_delivery_limit_single_node 方案对比 3 次投递后 delivery_limit 死信

运行入口是 examples/18-rabbitmq-consumer/run.sh，环境在 evidence/18/env.txt（RabbitMQ 3.12.1、amqp-client 5.22.0、PostgreSQL 16.15、OpenJDK 21.0.12.1），每个消费者进程的输出在 evidence/18/logs/，证明范围在 evidence/18/LIMITS.md。本次运行没有出现生产端 basic.nack，nack 后重发的路径只有代码，没有被实际触发。

结论边界

在单个 RabbitMQ 节点上，实验支持以下结论：publisher confirm 加持久化标志，能让已确认消息在 broker 平滑重启和 kill -9 之后保留；ack 不代表入队，需要 mandatory 与 return 处理配合；提交后再确认、再加上与副作用同事务的去重表，能让确认丢失导致的重投只施加一次副作用；有界预取把每个消费者的未确认数限制在设定值以内；重试队列让毒消息在有限次尝试后被隔离，不阻塞其他消息。

实验不支持以下结论：多副本持久性与集群分区下的行为（单节点，quorum 队列只有一个成员），掉电下已确认消息是否仍在（kill -9 不丢页缓存），以及任何吞吐数字。发布进程自身崩溃后，未确认消息如何从上游补齐，本篇没有覆盖；常见的做法是先把事件写进业务库，再由单独的进程发布，那属于事务与消息一致性的问题。

下一篇 19 延迟任务的领取与取消 处理另一类异步工作：任务要到某个时刻才能执行，执行者可能失联，取消请求可能和到期领取同时发生。那里同样允许重复的执行尝试，副作用仍然要单独去重。