RocketMQ之 CommitLog

1 消息存储概述

RocketMQ 的存储文件，放在 ${ROCKET_HOME}/store 目录下。

当生产者发送消息时，broker 会将消息存储到 commit 文件下，然后再异步的转存到 consumeQueue 以及 indexFile。

commitlog
消息主体以及元数据的存储主体。Producer 发送的消息就存放在 commitlog 里面.

consumeQueue
消息消费队列，引入的目的主要是提高消息消费的性能，由于 RocketMQ 是基于主题 topic 的订阅模式，消息消费是针对主题进行的，如果要遍历 commitlog 文件中根据 topic 检索消息是非常低效的。

indexFile
IndexFile（索引文件）提供了一种可以通过key或时间区间来查询消息的方法。

下图解释了，当写入 commitlog 时。commitlog, consumeQueue, indexFile 3 者的关系。

2 认识 commitLog

commitlog 是 RocketMQ 用于存储消息的文件. commitlog 具有以下特征

1. 顺序写入，随机读写。
2. 消息只要被写入 commitlog 那么该消息就不会丢失
3. 消息是非定长的
4. 每个文件大小默认 1GB
5. 文件的命名是以 commitlog 起始偏移量命名的

下面列出，commitlog 在磁盘上的表现

3 commitLog 写入流程

这个流程中，我们只需要关注 2 个点。

1. appendMessage()
2. 处理刷盘

appendMessage() 处理的逻辑是将消息写入堆外内存中，这里不过多关注。读者感兴趣可以自行查看源码，源码的位置 DefaultAppendMessageCallback#doAppend()

4 刷盘

4.1 同步刷盘

从使用者角度来看的同步刷盘流程

从程序角度来看的同步刷盘流程

1. CommitLog 将消息封装成 GroupCommitRequest, 并放入队列中。
2. 唤醒 GroupCommitServiceThread 线程
3. GroupCommitServiceThread 从 List 中获取 Request 并消费
4. 调用 MappedFile 将数据刷到磁盘上
5. 刷盘后，唤醒 CommitLog 所在线程

4.2 异步刷盘

从使用者角度来看的异步刷盘流程

transientStorePoolEnable 关闭时, 即 FlushRealTimeService 刷盘逻辑

基本流程：

代码位置：
CommitLog$FlushRealTimeService#run()

1. 每隔 500ms 执行一次 flush()
2. 每次刷盘时，需要凑够至少 4 页。当 2 次刷盘的时间间隔大于 10s 时，则忽略该条件，直接执行 flush()
3. 刷完盘之后，记录 checkPoint(刷盘点)

可配置参数

# false 使用 AQS 等待; true 使用 sleep 等待
flushCommitLogTimed=false
# 每隔 500ms 刷盘
flushIntervalCommitLog=500
# 每次刷盘时，至少4页
flushCommitLogLeastPages=4
# 间隔多久未刷盘，会强制刷盘
flushCommitLogThoroughInterval=1000*10

transientStorePoolEnable 开启时。RocketMQ 会通过 CommitRealTimeService 线程调用 FileChannel#write() 方法，将消息写入 cache. 然后唤醒 FlushRealTimeService 线程，由该线程执行 flush()

基本流程

代码位置
CommitLog$CommitRealTimeService#run()

1. 每隔 200ms 执行一次 write()
2. 每次执行 write() 时，需要凑够至少 4 页。当 2 次 write 的时间间隔大于 200ms，则忽略该条件，直接执行 write()
3. 调用 FileChannel#write()
4. 唤醒 FlushRealTimeService 线程
5. 由 FlushRealTimeService 执行 flush

# 每个 200ms 执行 write
commitIntervalCommitLog=200
# 每次执行 write 时，至少4页
commitCommitLogLeastPages
# 间隔多久未刷盘，会强制刷盘
commitCommitLogThoroughInterval=200

4.3 总结

RocketMQ 提供了 2 种刷盘方式, 可通过 flushDiskType 进行配置

1. flushDiskType=SYNC_FLUSH
   同步刷盘，调用 flush
2. flushDiskType=ASYNC_FLUSH & transientStorePoolEnable=false
   异步刷盘，调用 flush.
3. flushDiskType=ASYNC_FLUSH & transientStorePoolEnable=true
   异步刷盘, 调用 write，并随之调用 flush

5 参考

• rocketMQ 设计指南
• Java NIO 文件通道 FileChannel 用法及原理
• ByteBuffer详解