RocketMQ 的文件存储结构,CommitLog、ConsumeQueue 和 IndexFile
RocketMQ 的文件存储设计是其能够支撑超高并发和海量消息的核心。它采用了混合存储(所有 Topic 的消息写同一个文件)和顺序写、随机读的架构。
为了实现高效的读写,RocketMQ 的存储结构主要由三大核心文件组成:CommitLog、ConsumeQueue 和 IndexFile。
1. CommitLog(消息主体存储文件)
CommitLog 是 RocketMQ 真正的物理存储文件,所有的消息(无论哪个 Topic)到达 Broker 后,都会按到达顺序追加(Append)写入到 CommitLog 中。
- 存储机制: 绝对的顺序写。这种设计极大地提高了磁盘的写入吞吐量(顺序写磁盘的速度甚至能赶上写内存)。
- 文件大小: 每个 CommitLog 文件固定为 1GB(1073741824 字节)。
- 文件命名: 文件名长度为 20 位数字,表示该文件第一条消息的全局物理偏移量(Offset)。例如,第一个文件名为
00000000000000000000,如果该文件写满了 1GB,第二个文件名就叫00000000001073741824。 - 消息结构: 文件中存储的是变长的消息数据。包含了消息体(Body)、Topic 名称、队列 ID(QueueId)、消息属性(Properties)、Tags 等所有完整信息。
痛点引入: 既然所有 Topic 的消息都混在一个文件里,那消费者怎么按 Topic 消费呢?如果直接遍历 CommitLog 去找,效率无异于大海捞针。为了解决这个问题,引入了 ConsumeQueue。
2. ConsumeQueue(逻辑消费队列)
ConsumeQueue 是 CommitLog 的索引文件,相当于关系型数据库中的“非聚集索引”。它的作用是帮助消费者快速定位到自己订阅的 Topic 下的特定消息。
- 组织方式: 按照
Topic和QueueId分层建立目录。例如:$HOME/store/consumequeue/{topic}/{queueId}/。消费者正是通过这个文件来拉取消息的。 - 文件结构: 每个 ConsumeQueue 文件包含 30 万个条目(Entry),每个条目大小固定为 20 字节。因此每个文件约为 5.72 MB。
- 条目内容(20 字节的设计极其精妙):
- CommitLog Offset(8字节): 消息在 CommitLog 中的物理偏移量。
- Size(4字节): 消息的大小。
- Message Tag HashCode(8字节): 消息 Tag 的哈希值。
- Tag HashCode 的作用: 消费者在订阅时经常会使用过滤条件(如
TagA || TagB)。在 ConsumeQueue 中存储 Tag HashCode,Broker 在服务端就可以直接按 Hash 值过滤掉不匹配的消息,避免了将无效消息从 CommitLog 中读出带来的磁盘 IO 开销。
3. IndexFile(消息按键查询索引)
IndexFile 是为了提供按 Message Key 或 Message ID 查询消息而设计的。当我们需要在控制台(Dashboard)根据订单号、流水号等 Key 查找某条消息的具体内容时,就需要用到它。
- 文件命名: 以创建时的时间戳命名(例如:
20231010121212333)。 - 内部结构: 它的底层实现非常像 Java 中的
HashMap,是一个基于哈希槽和链表的数据结构,用于解决哈希冲突。 - 文件组成(固定大小约 400MB):
- Header(40字节): 包含文件的起始/结束时间、最小/最大物理偏移量、哈希槽数量、索引项数量等元数据。
- Hash Slot(哈希槽,500万个): 每个槽 4 字节,存储该 Hash 对应的最新的 Index Item 的序号。
- Index Item(索引项,2000万个): 每个条目 20 字节。
- Index Item 内容(20字节):
- Key Hash(4字节): Message Key 的哈希值。
- CommitLog Offset(8字节): 消息在 CommitLog 的物理偏移量。
- Timestamp(4字节): 消息存储时间与文件起始时间的差值(节省空间)。
- Next Index No(4字节): 发生 Hash 冲突时,指向前一个冲突项的序号(单向链表结构)。
4. 这三者是如何协同工作的?
了解了三者的结构,我们来看看一条消息从写入到被消费、被查询的全流程(这也是 RocketMQ 面试最常考的地方):
A. 消息写入(同步写,异步建索引)
- Producer 发送消息到 Broker。
- Broker 将消息顺序追加到
CommitLog中。只要写入 CommitLog 成功,就向 Producer 返回成功。 - Broker 后台有一个名为
ReputMessageService的异步线程,它会不断地读取 CommitLog 中新写入的数据。 - 该线程解析消息的 Topic、QueueId、Key 等信息,异步、实时地生成
ConsumeQueue和IndexFile的索引条目。
B. 消息消费(基于 ConsumeQueue)
- Consumer 发起拉取请求,带上自己的 Topic、QueueId 以及逻辑偏移量(Offset)。
- Broker 找到对应的
ConsumeQueue文件,定位到 20 字节的索引条目。 - 如果消费者配置了 Tag 过滤,Broker 首先对比条目中的
Tag HashCode,不匹配则跳过。 - 匹配成功后,取出
CommitLog Offset和Size。 - 拿着 Offset 和 Size 去
CommitLog中读取真正的消息主体,返回给 Consumer。
C. 消息查询(基于 IndexFile)
- 用户在控制台输入 Message Key(如订单号
Order123)。 - Broker 将
Order123进行 Hash 计算。 - 在
IndexFile中找到对应的 Hash 槽。 - 顺着 Hash 槽的链表(Next Index No)遍历,比对时间戳等信息。
- 找到匹配项后,取出
CommitLog Offset,去CommitLog中读出完整消息返回。
总结
RocketMQ 的文件存储设计完美诠释了“空间换时间”和“读写分离(物理写、逻辑读)”的理念:
- CommitLog 保证了极高的写入性能。
- ConsumeQueue 保证了高效的按 Topic 顺序消费。
- IndexFile 提供了灵活的业务键查询能力。
配合 Linux 的 PageCache 和 Mmap(内存映射) 技术,使得这套基于磁盘的存储系统能够拥有接近内存级别的读写效率。
右滑查看面试常问