系列导航：上一篇 16 服务发现中的旧地址 ｜ 本篇 17 ｜ 下一篇 18 消息队列接入与积压恢复

生产者收到一句“发送成功”，这句话到底承诺了什么？消息可能只是进了 broker 的内存队列，可能已经交给操作系统但还停在页缓存里，也可能已经被 fsync 刷到设备。三种状态在正常运行时看不出差别，进程崩溃、掉电或者磁盘出错时才会分开。消费端也有同样的问题：消费者读到了消息、处理了消息、提交了进度，这三件事的先后顺序决定了重启后丢消息还是重复处理。

本篇用一个单机日志队列把这些位置逐一拆开：分段追加日志、带长度与 CRC32C 的记录格式、启动时的恢复扫描、broker 管理的消费组进度，以及“确认—落盘—消费确认”三条彼此独立的线。代码在 examples/17-log-queue/，全部输出在 evidence/17/output.txt。

先划清边界：这是教学用的单机实现，没有复制，没有共识，没有多副本确认，也没有分区与再均衡。它最多只能说明“一块本地磁盘上的一个进程”在进程崩溃后能保住什么；单机磁盘损坏时数据就丢了，任何关于集群持久性的结论都不能从这里推出。

需求与保证

broker 只提供四个操作：发布、拉取、确认、查询状态。协议是一行一个请求的 TCP 文本协议，载荷是第 17、18 两章共用的合成订单事件（data/orders-events.jsonl，固定种子生成 3000 条，字段只有测试值）。

发布时，生产者指定要等待的确认级别：

级别 broker 回复时的状态 进程崩溃后 掉电后
ACCEPTED 已分配偏移，进入内存队列 可能不存在，偏移可能被重新分配 可能不存在
WRITTEN FileChannel.write 已返回，数据在页缓存 通常仍在（页缓存属于内核） 可能不存在
DURABLE FileChannel.force 已返回 存在 取决于设备是否如实执行刷新

表中“进程崩溃后”一列由本章实验观察，“掉电后”一列来自文档，本机没有验证（见后文 NOT_RUN 一节）。

broker 对外承诺的不变量只有下面几条：

  • 收到 DURABLE 回复的偏移，在进程崩溃并重启后存在，且内容与发布时一致。
  • 偏移从 0 开始连续，一经 DURABLE 确认就不再改变归属。
  • 消费者只能读到已经 force 的记录，即 durable 水位以下的记录。
  • 消费组的已提交偏移只有在持久化之后才回复 ACKED；未确认部分在连接断开后从已提交偏移重投，语义是至少一次。
  • 日志中间出现损坏时 broker 拒绝启动，不静默跳过任何记录。

broker 不承诺生产端去重（没有生产者编号与序号），也不承诺一个消费组可以并行消费：同一组同时只允许一个活动连接，第二个连接得到 ERR GROUP_BUSY。

规模假设是教学取值：单条载荷不超过 64 KiB，内存受理队列 1024 条，段文件默认 1 MiB，单个消费组单个消费者。这些数字只决定拒绝条件，不对应任何生产环境。

三条确认线

“受理”“落盘”“消费确认”是三件事，分别属于不同的持久状态。图 17-1 按时间画出一次 DURABLE 发布与一次消费，标出每个回复发生的位置和崩溃时各自的结果。

图 17-1：DURABLE 发布与消费确认的时序（已实现并经注入验证）

sequenceDiagram
    participant P as 生产者
    participant B as broker 连接线程
    participant W as 写线程
    participant D as 段文件
    participant C as 消费者
    P->>B: PUB DURABLE payload
    B->>B: 分配偏移 n，入内存队列（ACCEPTED 在此回复）
    B->>W: Pending(n)
    W->>D: write（WRITTEN 在此回复）
    Note over W,D: 注入点 written：此时崩溃，客户端无回复，记录仍在页缓存
    W->>D: force（一批只一次）
    Note over W,D: 注入点 forced：此时崩溃，记录已落盘，客户端无回复
    W-->>B: durable 水位推进到 n+1
    B-->>P: OK n DURABLE
    Note over P,B: 客户端 30s 读超时或连接断开 = 结果未知，只能重试
    C->>B: FETCH g 10 500（只读水位以下）
    B-->>C: MSG n payload
    C->>C: 施加副作用
    C->>B: ACK g n
    B->>B: 写 g.offset.tmp → force → rename → fsync 目录
    B-->>C: ACKED n+1

生产者丢失回复时无法区分“写线程还没写”与“已经落盘只是回复没到”，因此它只能重试，重试又可能产生重复记录。crash_forced_before_reply_dup 观察到了这种重复：force 之后、回复之前停机，生产者重试后同一个 message_id=evt-000011 出现在偏移 [10, 11]。偏移去重挡不住这种重复，消费端需要按业务标识去重，第 18 章再处理这个问题。

ACCEPTED 的偏移更不稳定。accepted_offset_not_stable 在受理后、写入前注入停机：生产者已经收到 OK 5 ACCEPTED，重启后偏移 5 被分配给另一条消息（evt-000201），原载荷在日志中不存在。把 ACCEPTED 的偏移写进下游数据库当作引用，就会指向一条别的消息。

记录格式与持久状态

每条记录由 16 字节头和载荷组成：长度 u32、CRC32C u32、偏移 u64。CRC 覆盖长度、偏移和载荷，因此长度字段或偏移字段被改动也能检测出来。CRC32C 由 JDK 的 java.util.zip.CRC32C 计算；PostgreSQL 的 WAL 采用同一类校验，每条记录都有 CRC-32C，并在崩溃恢复时检查（PostgreSQL 16 §30.1）。

图 17-2：broker 的状态归属与持久化位置

1
2
3
4
5
6
7
8
9
           内存（进程退出即失去）                           磁盘（<dir>/）
┌──────────────────────────────────────┐    ┌──────────────────────────────────────────────┐
│ nextSeq        下一个待分配偏移        │    │ log/00000000000000000000.log  段 0            │
│ queue[1024]    ACCEPTED 的待写记录     │──▶ │ log/00000000000000000123.log  段 1（文件名=   │
│ durable 水位   force 已覆盖的偏移上界   │    │                               首条记录偏移）  │
│ 偏移→位置索引   启动时由扫描重建         │    │   每条：len | crc32c | offset | payload       │
│ 连接游标        每连接、每组的投递位置    │    │ groups/<g>.offset             已提交偏移       │
└──────────────────────────────────────┘    │ groups/<g>.offset.tmp         仅在写入途中存在 │
       写线程独占写；读者只读水位以下            └──────────────────────────────────────────────┘

内存索引没有持久化，启动扫描时按顺序读取每条记录重建，代价与日志大小成正比。这是“单文件 vs 分段+索引”取舍里的折中：分段让滚动、删除旧数据、定位文件都只涉及文件名；持久化的稀疏索引能让启动不必读完全部数据，但索引自身也会在崩溃时写坏，需要校验或可重建。教学实现选择每次重建，代价是启动时间随日志线性增长，boundary_segments_and_rejects 用 4096 字节的段验证了跨 4 个段文件读回 100 条记录。

段滚动时先 force 旧段，再创建新段并对目录调用 fsync。Linux 的 fsync 手册说明，对文件 fsync 不保证包含它的目录项也已落盘，需要对目录的文件描述符再做一次（fsync(2)）。在 Java 里，以只读方式打开目录得到的 FileChannel 在 Linux 上可以调用 force。这一步在本机无法通过故障注入证伪，因为进程崩溃不会丢掉目录项；它的依据只有手册。

写线程与组提交

所有写入由一个写线程完成。它从内存队列一次取出最多 256 条，逐条 write，再整批 force 一次，然后统一回复 DURABLE。下面是 src/Broker.java 中组提交部分的主干（省略了注入点）：

1
2
3
4
5
6
7
8
9
10
Pending first = queue.take();
batch.clear(); batch.add(first); queue.drainTo(batch, 255);
for (Pending p : batch) {
    log.maybeRoll();
    log.append(p.offset, LogStore.encode(p.offset, p.payload));
    if (p.level == Level.WRITTEN) p.done.complete(null);
}
long last = batch.get(batch.size() - 1).offset;
doForce(last + 1);                                // 一批只 force 一次，并推进 durable 水位
for (Pending p : batch) if (p.level == Level.DURABLE) p.done.complete(null);

FileChannel.force 的 Javadoc 只对本地存储设备作保证：方法返回时，自通道创建或上次调用以来的修改都已写入该设备；文件不在本地设备上时没有这种保证（Java SE 21 FileChannel）。实现用的是 force(false)，只要求文件内容，文件长度等元数据是否一并刷新由操作系统决定，Javadoc 明确写着这一点未规定。

写线程遇到 IO 异常时直接停机，不再受理。fsync 失败之后，页缓存里哪些数据被丢弃、哪些仍标记为脏，取决于内核版本与文件系统；在这种状态下继续 force 并回复成功，可能把已丢失的数据报告为持久。停机后重新扫描磁盘，是本实现能给出的最保守处理。

逐条 force 与组提交的对比结果如下，两组都是 4 个并发生产者、共 400 条 DURABLE 发布，之后 kill -9 并重启校验：

方案 force 次数 每次 force 平均 总耗时 kill -9 后已确认偏移
组提交（normal_group_commit） 228 271µs 186ms 400/400 存在且一致
逐条 force（compare_force_each） 400 210µs 156ms 400/400 存在且一致

稳定的结论只有 force 次数：组提交把 400 次降到 228 次，批次大小取决于线程调度，每次运行都不同。耗时一列不能拿来比较两种方案，本机一次 force 只要零点几毫秒，这个量级下的差异落在噪声里；这块存储可能有带掉电保护的缓存，也可能没有如实执行刷新，本机无法区分。组提交的收益在刷新代价高的设备上才会显现，代价是同一批里先到的请求要等最后一条写完才得到回复。Kafka 的设计文档走的是另一条路：数据写入文件系统但不一定刷盘，持久性靠副本重新同步，理由是每次写都 fsync 会让性能下降两到三个数量级（Kafka 3.9 Design）。那条路以复制为前提，单机日志不能照搬。

恢复扫描：截断尾部，拒绝中间损坏

启动时，LogStore.open 按文件名顺序扫描每个段，逐条检查四件事：头部是否完整、长度是否在 1…65536 之内、载荷是否装得下、CRC 与偏移连续性是否成立。难点在于遇到第一条无效记录之后怎么办。

写入中途崩溃只会损坏最后一段的尾部：追加日志只在末尾写，残缺记录之后不会再有有效记录。中间损坏不一样，它之后还有完整的记录，其中可能有已经回复过 DURABLE 的偏移。判断方法是从无效位置的下一个字节开始，逐字节寻找任意一条 CRC 有效的记录：

1
2
3
4
5
6
7
long later = last ? findValidAfter(ch, pos + 1, fileSize) : -2;
if (!last || later >= 0) {                 // 非最后一段，或其后仍有有效记录：中间损坏
    r.status = "CORRUPT"; ...
    closeAll(); return r;                  // Broker 输出 REFUSE_START，退出码 3，不改文件
}
r.status = "TRUNCATED_TAIL"; ...           // 残缺尾部：截断并报告
if (repair) { ch.truncate(pos); ch.force(true); }

图 17-3：恢复扫描对一条无效记录的判定（前置条件与拒绝路径）

1
2
3
4
5
6
7
8
读到位置 pos 的记录
  │ 头部完整？长度合法？载荷装得下？CRC 相符？偏移 = 期望值？
  ├─ 全部是 ──────────────▶ 记入索引，pos 前进，期望偏移 +1
  └─ 任一否
       ├─ 不在最后一段 ─────────────────────────▶ CORRUPT：拒绝启动（exit 3），不修改文件
       └─ 在最后一段
            ├─ pos 之后能找到 CRC 有效的记录 ──────▶ CORRUPT：拒绝启动（exit 3），不修改文件
            └─ 找不到 ───────────────────────────▶ TRUNCATED_TAIL：截断到 pos、force、报告字节数与偏移

“逐字节寻找”这一步专门对付长度字段损坏。若中间某条记录的长度被改成 60000，朴素的扫描会认为它“载荷装不下”，再把它当作残缺尾部截断，后面的 20 条有效记录随之消失。mid_corruption_refused 的第二部分构造了这个场景：恢复扫描在位置 4648 遇到 short-payload，又在 4806 找到有效记录，于是判定 CORRUPT 并以退出码 3 拒绝启动。第一部分翻转了第 20 条载荷里的一个字节，得到 reason=crc-mismatch ... expectedOffset=20 validRecordAt=3244，文件大小没有变化。

遇到中间损坏时可以拒绝启动，也可以把坏记录挪到隔离区后继续服务。本实现选拒绝启动。坏记录所在的偏移可能已经被确认为 DURABLE，自动跳过会让偏移序列出现空洞，消费者无法区分“这个偏移从未存在”与“这个偏移被跳过了”；而隔离需要额外的元数据格式、修复工具和运维流程，教学实现不承担这部分。拒绝启动牺牲的是可用性，换来的是不会把损坏悄悄交给消费者。

这条规则也有盲区。如果恰好是最后一条记录的载荷被翻转，它之后没有任何有效记录，扫描无法区分“写到一半”和“写完后被损坏”，会按残缺尾部截断。若这条记录已经被确认为 DURABLE，截断就丢了一条已确认的数据。要消除这个盲区，需要额外持久化 durable 水位，或依赖副本比对。PostgreSQL 面对的是同一类问题：文档指出掉电可能让一次写入只完成部分 512 字节扇区，它在修改数据页之前先把整页镜像写入 WAL，恢复时用镜像修复这些 partially-written pages，WAL 记录本身则靠 CRC-32C 判断内容是否正确（PostgreSQL 16 §30.1）。

消费进度与重投

消费组进度由 broker 管理，存成一个十进制文本文件。更新步骤是：写 g.offset.tmp，force，用 ATOMIC_MOVE rename 成 g.offset，再 fsync groups/ 目录，最后回复 ACKED。rename 替换已存在目标时是原子的，其他进程不会看到目标文件缺失的时刻（rename(2)）。

group_offset_crash_points 在两个位置注入停机。rename 之前停机，重启时删除残留的 .tmp（GROUP_TMP_REMOVED file=g.offset.tmp），进度仍是上一次的 20，消费者从 20 重新读取；rename 之后、目录 fsync 与回复之前停机，进度已经是 40，尽管客户端没有收到 ACKED。后一种情况对消费者无害：它以为没提交，重新确认时 broker 发现值不大于已提交值，不回退。

消费者的副作用写入 PostgreSQL 库 b2_17 的 applied 表，偏移是主键，insert ... on conflict (log_offset) do nothing 让重复投递不产生第二行。consumer_redelivery_dedup_by_offset 让消费者每 10 条确认一次，在处理完第 25 条（偏移 24）后停机，此时已提交的是 20：

1
2
CONSUMER_HALT seen=25 lastOffset=24 (applied, not acked)
CONSUMER_DONE firstOffset=20 seen=40 applied=35 skippedDuplicates=5 | rows=60 distinctOffsets=60

重启后从 20 开始重投，偏移 20 到 24 被跳过，最终 60 行对应 60 个偏移。负向对照 negative_consumer_without_dedup 去掉主键约束，同样的崩溃留下 65 行，偏移 20 到 24 的副作用各施加了两次。这种顺序在 Kafka 设计文档里称为“先处理、后保存位置”，对应至少一次（Kafka 3.9 Design）。

进度由谁保存是另一项取舍。broker 管理进度，消费者实现简单，broker 能据此决定重投起点，但进度与副作用分处两个存储，中间的窗口只能靠去重弥补。消费者自管偏移，把偏移和副作用写进同一个数据库事务，就能让二者一起成功或一起失败，代价是 broker 不知道消费到哪里，换消费者或排查积压时要去读消费者的库。本实验的 applied 表其实已经具备自管偏移的条件，只是没有用它来决定拉取起点。

进程崩溃不等于掉电

kill -9 和 Runtime.halt 只结束进程，内核里的页缓存不受影响。crash_written_before_force 在 write 之后、force 之前停机，客户端没有收到回复，重启后偏移 10 的记录仍然存在（offset10PresentAfterRestart=true），恢复扫描结果是 status=OK records=11。同样的道理，kill9_durable_acked_survive 在发布途中 kill -9，3298 个已确认偏移全部找回，这个结果对 WRITTEN 级别同样成立，所以它不能区分 force 有没有起作用。

真正区分 WRITTEN 与 DURABLE 的是掉电或内核崩溃。Linux 手册说明 fsync 会把文件在内核中修改过的数据传到设备，包括写穿或刷新磁盘缓存；PostgreSQL 文档补充，控制器和磁盘的写回缓存是易失的，有些 SSD 默认不执行刷新命令，可靠性最终取决于整条存储链路。本机是 docker 容器，没有 dmsetup，也没有可加载的内核模块，无法用 dm-flakey 或 dm-log-writes 丢弃未刷新的写入，也无法给虚拟机强制断电，因此 power_loss_unforced_data 记为 NOT_RUN。

为了让“确认早于持久化”这个错误能被观察到，negative_early_durable_ack 用了一个标明为模型的注入：--sim-power-loss 在停机前把活动段截回最近一次 force 时的长度，模拟页缓存丢失。错误实现 --ack early 在 write 之后先回复 DURABLE 再 force，在回复之后、force 之前停机：

1
2
model=simulated-page-cache-loss early: acked=16 missingAcked=1 ... offset15Acked=true
                                 | correct: acked=15 missingAcked=0 ... offset15Acked=false

错误实现丢了 1 个已确认偏移；正确实现在 write 之后、force 之前做相同的注入，偏移 15 从未被确认，已确认偏移没有丢失。这个对照说明的是确认顺序带来的窗口，不能当作掉电实验的结果。

只校验长度的恢复

另一个负向对照是去掉 CRC 检查的恢复（--recovery length-only）：只要头部完整、长度装得下就算有效。注入点 extend 写入记录头后，在记录末尾写一个字节，使文件长度覆盖整条记录、载荷部分是全零空洞。这个注入构造的是“长度已经覆盖整条记录、内容却是零”的字节形状；真实掉电能否留下这种形状取决于文件系统的日志模式，本机没有验证。

1
2
crcRecovery: RECOVERY status=TRUNCATED_TAIL ... truncatedBytes=157 truncatedAtOffset=20 reason=crc-mismatch
lengthOnly: records=21 offset20Payload=141xNUL deliveredToConsumer=true (expectBug)

带 CRC 的恢复截断了 157 字节；只看长度的恢复接受了偏移 20，把 141 个 NUL 字节交给消费者。对中间翻转的那份日志，它照常启动，mismatched=1：一个已确认偏移的内容被静默改变。

用例与证据

用例 类型 观察值（摘自 evidence/17/output.txt）
normal_group_commit / compare_force_each 正常 / 方案对比 400/400 已确认偏移在 kill -9 后存在且一致；force 228 次 vs 400 次
boundary_segments_and_rejects 边界 4 个段文件；超长载荷 ERR TOO_LARGE；空拉取 300ms 后返回 0 条；越界确认 ERR NOT_DELIVERED；同组第二连接 ERR GROUP_BUSY
kill9_durable_acked_survive 崩溃恢复 3298 个已确认偏移，缺失 0、内容不符 0
crash_partial_write_truncated 崩溃恢复 TRUNCATED_TAIL 79 字节，偏移 30 起截断，0…29 完整
crash_written_before_force 崩溃恢复 客户端无回复，记录仍在（进程崩溃不丢页缓存）
crash_forced_before_reply_dup 结果未知 重试后同一事件位于偏移 10 与 11
accepted_offset_not_stable 确认分级 OK 5 ACCEPTED 后停机，偏移 5 重新分配
mid_corruption_refused 损坏 字节翻转与长度损坏均 CORRUPT、退出码 3、文件不变
negative_lengthonly_accepts_tail / _flip 负向对照 全零记录与被改内容交给消费者
negative_early_durable_ack 负向对照（模型） 早确认丢 1 个已确认偏移，正确实现丢 0 个
consumer_redelivery_dedup_by_offset / 无去重对照 重投 从 20 重投，去重跳过 5 条；无去重 65 行
group_offset_crash_points 崩溃恢复 rename 前停机进度为 20，rename 后停机进度为 40
power_loss_unforced_data 掉电 NOT_RUN，需要虚拟机断电或 dm-flakey

运行入口是 examples/17-log-queue/run.sh，环境在 evidence/17/env.txt（OpenJDK 21.0.12.1、PostgreSQL 16.15、4 核、ext4 上的 docker 容器），每个 broker 与消费者进程的输出在 evidence/17/logs/。证明范围与时序敏感点见 evidence/17/LIMITS.md。

结论边界

实验支持的结论限于单机、单进程崩溃：收到 DURABLE 的偏移在 kill -9 与各个注入点之后存在且内容一致；写到一半的尾部被检测、截断并报告；中间损坏导致拒绝启动，而不是被跳过；消费者在副作用与确认之间崩溃会被重投，按偏移去重可以吸收这部分重复。ACCEPTED、WRITTEN、DURABLE 三个确认点和消费确认，各自对应不同的持久状态，不能互相替代。

实验不支持以下结论：掉电或内核崩溃下 DURABLE 是否成立（NOT_RUN），目录 fsync 是否必要（没有能丢目录项的环境），以及任何吞吐与延迟判断。此外，这个队列没有副本：Kafka 的承诺以“至少一个同步副本存活”为前提（Kafka 3.9 Design），本实现连这个前提都没有，单机日志实验得出的任何数字都不能用来推断集群持久性。

下一篇 18 消息队列接入与积压恢复 换成真实的 RabbitMQ 节点，用同一份订单事件检验客户端一侧的契约：publisher confirms 在什么时候回来，消费确认丢失时副作用怎样只发生一次，毒消息怎样隔离，积压时预取怎样限制未确认数量。