技术小册

主题

RocketMQ 消息存储与持久化 ​

一句话：所有 Topic 共用一份 CommitLog 顺序追加，每个 Queue 用 20 字节的 ConsumeQueue 条目指向 CommitLog；刷盘策略决定宕机丢多少，位点存储决定重启从哪继续。

适用版本：4.9 / 5.3 / 5.5。存储目录结构与刷盘参数基本一致，5.x 增加了 RocksDB 索引等可选实现。

为什么需要 ​

消息中间件要同时满足三件事：写入吞吐高、单机容量大、重启不丢。随机写盘达不到吞吐要求，所以必须顺序写；但按 Topic 各写一份日志，Topic 一多就会遇到文件句柄和刷盘竞争。RocketMQ 的选择是把所有消息写进同一份 CommitLog，再为消费侧建轻量索引。

这套设计还有一个隐性收益：ConsumeQueue 损坏可以丢，CommitLog 是唯一数据源，重建索引不影响消息本身。理解这一点，才能判断哪些「丢数据」是真丢，哪些只是索引落后。

核心机制 ​

磁盘上有什么 ​

text
${storePathRootDir}/
├── commitlog/
│   └── 00000000000000000000        # 文件名 = 该文件起始物理偏移，默认 1 GiB 滚动
├── consumequeue/
│   └── topic-a/0/00000000000000000000   # 每 Topic 每 Queue 一组索引文件
├── index/                          # 按 key 建的反查索引（queryMsgByKey 依赖它）
├── config/
│   ├── consumerOffset.json         # 集群消费位点
│   ├── delayOffset.json            # 4.x 延迟消息各级别进度
│   ├── topics.json
│   └── subscriptionGroup.json
├── checkpoint                      # 各刷盘/写入指针
├── abort                           # 非正常退出时存在，重启触发恢复
└── lock

CommitLog：顺序写与消息格式 ​

每条消息在 CommitLog 中是一段自描述记录，依次包含：总长度、magic code、body CRC、QueueId、物理偏移、Queue 内逻辑偏移、生产时间、存储时间、重投次数、属性等。写入用 mmap 映射文件，追加后更新 wrotePosition；flushedPosition 只在 fsync 后前进。

恢复时先看 abort 文件是否存在。存在说明上次非正常退出，从最后一个 CommitLog 文件尾部向前扫描，校验 magic code 与长度，丢掉写了一半的尾部记录，得到准确的 maxOffset。

ConsumeQueue：每条 20 字节 ​

字段长度作用
CommitLog 偏移8B消息在 CommitLog 中的物理位置
消息长度4B读取完整消息所需的字节数
Tag hash8B在索引层先过滤，命中后才回 CommitLog 校验原始 Tag

消费时按 logicOffset × 20 定位索引条目，拿到 CommitLog 物理偏移和长度，再读完整消息。Tag hash 让 Broker 能在索引层先过滤掉不匹配的消息，减少无效的 CommitLog 读。

ConsumeQueue 由 dispatch 线程在消息写入后异步构建，起点是 reputFromOffset。Broker 重启时该值从 ConsumeQueue 最后一条反推；如果索引落后或损坏，dispatch 会从 CommitLog 重扫补齐。因此 ConsumeQueue 是可重建的派生数据，不是数据源。

三层 offset，只有一层需要持久化 ​

层含义是否持久化
CommitLog offset消息在 CommitLog 文件中的字节位置不需要，由文件本身决定
ConsumeQueue offset某 Queue 内第几条消息不需要，文件大小 ÷ 20 即最大条数
Consumer offset某消费组在某 Queue 消费到第几条需要，决定重启后的起点

消费位点存在哪 ​

集群消费（CLUSTERING）：Broker 端维护。Consumer 提交后先更新内存 offsetTable，每 5s 刷到 config/consumerOffset.json；重启时从该文件恢复。

系统 Topic OFFSET_MOVED_EVENT 不是位点备份。它只在 Broker 发现某个消费位点需要校正时写入一条事件，通知相关客户端调整拉取位置。把它当成「json 损坏后从 Topic 回放恢复位点」是错的——集群位点的权威存储就是 consumerOffset.json，丢了只能按 consumeFromWhere 策略重新选择起点。

广播消费（BROADCASTING）：位点不存 Broker，由客户端 LocalFileOffsetStore 每 5s 写到本地：

~/.rocketmq_offsets/{clientId}/{consumerGroup}/offsets.json

进程重启会先读这个文件，所以正常情况下是接着上次的位置，不是「从最新消息开始」。只有本地文件不存在或读取失败时，才按 consumeFromWhere（默认 CONSUME_FROM_LAST_OFFSET）从最新开始。换机器、清缓存、容器无持久卷，都会表现为「位点丢了」。

刷盘与可靠性 ​

flushDiskType行为宕机丢失窗口
ASYNC_FLUSH（默认）写入 PageCache 即返回，后台每 500ms 刷盘最多约 500ms 的写入
SYNC_FLUSH每条消息等 fsync 完成才返回已确认的消息不丢，但吞吐显著下降

异步刷盘 + 同步复制是常见组合：本地不承担 fsync 延迟，数据安全靠 Slave 副本。只开异步刷盘、又没有可用 Slave 时，机器掉电会丢最后一段。

过期清理与磁盘水位 ​

  • fileReservedTime=72：文件保留 72 小时。
  • deleteWhen=04：每天 04:00 触发过期文件删除。
  • 磁盘水位：使用率超过 75% 开始删除过期文件；超过 85% 强制清理（diskSpaceCleanForciblyRatio）；超过 90% 拒绝写入（diskSpaceWarningLevelRatio）。

清理以整文件为单位，不会删到文件中间；ConsumeQueue 的清理以 CommitLog 的最小 offset 为界，索引不会先于数据被删。

实战场景 ​

一、maxOffset 正常但消费看不到新消息：dispatch 落后 ​

现象：大促期间某 Topic 在控制台显示 Broker Offset（maxOffset）持续增长，但消费组的 Consumer Offset 停在两小时前的位置，界面上“积压=0”且没有任何报错；消费者日志里拉取请求正常返回空。运维先怀疑消费者线程池，重启消费端后毫无变化。

排查：ConsumeQueue 是 dispatch 线程异步构建的派生数据，落后于 CommitLog 时消费侧就只能看到旧索引。

定位层取证据判定
索引是否落后reputFromOffset 与 CommitLog flushedPosition/wrotePosition 的差值差值持续增大即 dispatch 跟不上
消费读的是索引消费按 logicOffset × 20 定位，索引没建出来就“无消息”表现为不报错的卡住
磁盘 IOiostat 的 util 与 await；CommitLog 顺序写是否与其他读争抢写放大挤占 dispatch
消息大小分布单条大小与 batch 数量；是否大量超大消息大报文让 dispatch 单位时间处理条数骤降
索引损坏Broker 重启日志里是否有重建（从 CommitLog 重扫补齐）重建期间消费滞后属预期
Topic/Queue 数量是否创建了异常多的 Queue（每 Queue 一套索引文件）小文件过多导致随机读放大

对症做法：

text
1. 先切断来源：限制超大消息（单条 > 128 KiB 走对象存储 + 引用），
   降低 Topic 数量与每 Topic 的 Queue 数，减少索引文件规模。
2. 存储层：CommitLog 与 ConsumeQueue 尽量同盘顺序写，避免与高随机读业务混部；
   写盘成为瓶颈时加盘或升 SSD，不要靠重启解决。
3. Broker 重启会让 dispatch 从 CommitLog 重扫补齐，期间消费滞后，
   要预估补齐时长并提前通知下游，不要在高峰重启。
4. 指标补上：dispatch 落后字节数、reputFromOffset 与写入位的差值、
   ConsumeQueue 文件数与大小；这三条比“消费 TPS”更早暴露问题。
properties
# 相关存储参数（以当前版本默认值为准，改动前压测）
mappedFileSizeCommitLog=1073741824        # 单文件 1 GiB，顺序写
deleteWhen=04
fileReservedTime=72

复盘要点：判据是“ConsumeQueue 构建延迟（写入位 - reputFromOffset）P99 应显著小于消费拉取周期”，并把“maxOffset 增长而 consumerOffset 不动”做成一条专门的告警——它是这类静默滞后的唯一信号。索引是可重建的派生数据，CommitLog 才是数据源；这条区分决定了处置方向是“补索引/降写入”，不是“恢复索引文件”。

二、磁盘到 85%，未消费的消息被强制清理 ​

现象：一个消费组因为下游数据库故障停了 3 天。恢复后发现部分消息永久丢失，Broker 日志显示 disk is used more than diskSpaceCleanForciblyRatio, delete file；deleteWhen=04 的定时删除并没有触发这么早，是水位强制清理干的。

排查：清理以整文件为单位，但水位阈值可以打破保留时间。

定位层取证据判定
当前水位磁盘使用率与 Broker 拒写/清理日志时间点85% 触发强制清理、90% 拒写
保留策略fileReservedTime=72（72 小时）停 3 天恰好跨过保留期，两者叠加致死
删除单位被删文件的 storeTimestamp 与消息写入时间整文件删除，不会删到中间
索引与数据边界ConsumeQueue 清理以 CommitLog 最小 offset 为界索引不会先于数据被删
消费进度该组 consumerOffset 与已删文件的 offset 区间消费位点落在已删区间即确认丢失
副本因素Slave 是否同盘同策略主备同时清理则没有恢复余地

对症做法：

properties
# 1. 水位告警要早于删除动作：75% 是删除起点而不是安全线
diskSpaceWarningLevelRatio=0.90
diskSpaceCleanForciblyRatio=0.85
# 运维口径：70% 预警、75% 扩容或清理、85% 已进入数据风险区

# 2. 保留时间与最慢消费者的恢复窗口对齐
fileReservedTime=168          # 关键业务从 72h 提到 7 天（按盘量校准，经验值需实测）
text
3. 优先级顺序：扩消费能力 > 扩盘/加 Broker > 延长保留时间。
   延长保留只是把风险推后，盘量不变时最终会更快触到水位。
4. 消费组停摆要有“位点接近清理线”的告警：
   比较各组 consumerOffset 与 Broker 最小可用 offset，差值超过阈值即叫人。
5. 不要手工删 commitlog 文件；要清理用 mqadmin deleteTopic 或调保留时间，
   让 Broker 自己按整文件滚动删除。

复盘要点：判据是“每个消费组的位点与最小可用 offset 之间留有多少小时的缓冲”，这个数字应该出现在容量周报里，而不是等事故。消息中间件的“不丢”承诺受物理盘约束：ASYNC_FLUSH 已有约 500ms 的宕机窗口，再叠加强制清理，可靠性要按最坏情况评估，见 高可用与主从复制。

三、容器化后广播位点丢失，每天全量重放 ​

现象：配置刷新类 Topic 用广播消费（BROADCASTING），每台实例都要收到全量消息。迁到 K8s 后出现两个相反症状：发布当天所有实例重放了近 6 小时的全部配置消息；另一次重启后部分实例完全没收到刚发布的配置。

排查：广播位点不存 Broker，存在客户端本地磁盘。

定位层取证据判定
消费模式MessageModel 是 CLUSTERING 还是 BROADCASTING广播位点由 LocalFileOffsetStore 管
位点文件是否存在~/.rocketmq_offsets/{clientId}/{consumerGroup}/offsets.json无文件即按 consumeFromWhere 选起点
clientId 是否稳定容器里 clientId 是否每次变化（默认含 IP/PID）变化后读不到旧目录，等价于全新消费者
是否有持久卷容器重建后该路径是否保留无 volume 就是每次从零开始
起点策略CONSUME_FROM_LAST_OFFSET / CONSUME_FROM_FIRST_OFFSETFIRST 会全量重放，LAST 会跳过未消费的
刷盘周期本地每 5s 写一次崩溃窗口内最多重复 5s 的消息

进程重启本来会先读本地文件接着消费，“重启从最新开始”只在本地位点文件缺失或读取失败时成立——换机器、清缓存、容器无持久卷都会表现为位点丢失，两种现象（重放与漏消费）取决于起点策略。

对症做法：

yaml
# 1. 给本地位点目录挂持久卷，并固定 clientId
volumeMounts:
  - { name: rmq-offsets, mountPath: /home/admin/.rocketmq_offsets }
volumes:
  - name: rmq-offsets
    persistentVolumeClaim: { claimName: rmq-offsets-{{ instanceName }} }
java
// 2. clientId 用稳定的实例名（StatefulSet 序号），不要依赖容器 IP
consumer.setInstanceName("config-refresh-" + podOrdinal);
consumer.setConsumerGroup("config-refresh-broadcast");
consumer.setMessageModel(MessageModel.BROADCASTING);
// 首次上线的起点要显式决定，不用默认值碰运气
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
text
3. 更稳的方向是改成集群消费 + 每实例一个消费组（组名带实例标识），
   把位点交给 Broker 的 consumerOffset.json 管理；
   或者把“配置类广播”换成长轮询拉取（客户端定时拉全量 + 版本号比较），
   天然不依赖位点。
4. 广播消费的处理逻辑必须幂等：同一配置消息可能被重复投递。

复盘要点：判据是“实例重启后收到的第一条消息 offset 是否等于重启前记录的 offset”，这条可以作为部署自检的一部分打印出来。广播与集群的差别不只在于“谁收到”，还在于“位点归谁管”：集群位点的权威存储是 Broker 的 consumerOffset.json，OFFSET_MOVED_EVENT 只是位点校正事件，不能当备份用。

最小上手 ​

bash
# 1. 看存储目录（路径取 broker.conf 的 storePathRootDir）
find /tmp/rmq-store -maxdepth 3 -type d | sort
ls -lh /tmp/rmq-store/commitlog | head
ls -lh /tmp/rmq-store/consumequeue/demo/0 | head

# 2. 发几条消息后看队列 offset
sh bin/mqadmin topicStatus -n 127.0.0.1:9876 -t demo
# 关注 minOffset / maxOffset / lastUpdateTimestamp

# 3. 看消费组位点与积压
sh bin/mqadmin consumerProgress -n 127.0.0.1:9876 -g cg-demo
# 输出每队列的 broker offset、consumer offset、diff

# 4. 看 Broker 运行态（磁盘、PageCache、刷盘）
sh bin/mqadmin getBrokerRuntimeInfo -n 127.0.0.1:9876

判断方法：

  • topicStatus 的 maxOffset 在涨、consumerProgress 的 diff 不降 → 消费慢，不是存储丢。
  • maxOffset 不涨但 Producer 说发送成功 → 查是否发到了别的 Queue/Topic，或 Broker 重启后路由未刷新。
  • consumerProgress 的 consumer offset 大于 broker offset → 位点异常，通常是重置位点或换组导致。

生产动作 ​

  • 按业务定 flushDiskType：支付、订单用 SYNC_FLUSH 或 ASYNC_FLUSH + SYNC_MASTER；日志、埋点用默认异步刷盘。
  • fileReservedTime 与消费速度匹配：积压超过保留时间，消息会在被消费前删除。积压场景先扩消费，再谈保留时长。
  • 磁盘水位必须告警：75% 是删除起点，不是安全线。到 85% 会强制清理，可能删掉还没消费的过期数据；到 90% 直接拒写。
  • 不要手工删 commitlog 文件。要缩容或清理，用 mqadmin deleteTopic 或调整保留时间，让 Broker 自己滚动删除。
  • 监控 dispatch 落后：ConsumeQueue 构建跟不上时，maxOffset 正常但消费看不到新消息，表现为「消费卡住但没报错」。
  • 换机器/容器化时持久化广播位点目录，否则每次发布都从最新位点开始，等于漏消费。

边界 ​

  • ConsumeQueue 可重建，CommitLog 不可。索引损坏不用慌；CommitLog 文件被删或被截断，消息就真没了。
  • OFFSET_MOVED_EVENT 不是备份。它只是位点校正事件，不能用来恢复 consumerOffset.json。
  • 广播消费位点在本地磁盘，不是内存。说「重启从最新开始」只在本地位点文件缺失时成立。
  • 异步刷盘不保证零丢失，即使 Broker 返回成功；要零丢失必须同步刷盘或等副本确认。
  • mmap 不是零拷贝读取。RocketMQ 写入靠 mmap 减少一次拷贝，读取仍要经过 PageCache；Kafka 的 sendfile 在纯转发场景拷贝次数更少，两者优化目标不同。

讲清楚 ​

RocketMQ 的存储是一份 CommitLog 加多份 ConsumeQueue。所有 Topic 的消息按到达顺序追加到同一个 CommitLog，单文件 1 GiB 滚动；每个 Queue 有一条 20 字节的索引，记录消息在 CommitLog 的物理偏移、长度和 Tag hash。消费时先读索引再读日志，多一次查找，换来 Topic 数量不受文件句柄限制。

三层 offset 里，前两层都不用存：CommitLog 的文件名就是起始偏移，ConsumeQueue 的文件大小除以 20 就是条数。真正要持久化的只有消费位点。集群消费存在 Broker 的 consumerOffset.json，每 5 秒刷一次；OFFSET_MOVED_EVENT 只是位点校正事件，不是备份。广播消费把位点写在客户端本地的 ~/.rocketmq_offsets，也是每 5 秒刷，所以重启通常能续上，只有本地文件丢了才从最新开始。

可靠性由刷盘和副本共同决定。默认异步刷盘最多丢 500 毫秒左右的写入；同步刷盘不丢但慢。磁盘到 75% 开始删过期文件，85% 强制清理，90% 拒写。排障时先分清是「消息没写进来」还是「消费没跟上」，topicStatus 和 consumerProgress 一看就知道。

把一门技术讲透到既懂也会用

技术小册 · 持续更新