基于本文回答

播面 播面

文解图解播客视频,多维讲透八股文
0
评论

RocketMQ 的文件存储结构,CommitLog、ConsumeQueue 和 IndexFile

知识点图片

RocketMQ 的文件存储设计是其能够支撑超高并发和海量消息的核心。它采用了混合存储(所有 Topic 的消息写同一个文件)和顺序写、随机读的架构。

为了实现高效的读写,RocketMQ 的存储结构主要由三大核心文件组成:CommitLogConsumeQueueIndexFile


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 下的特定消息。

  • 组织方式: 按照 TopicQueueId 分层建立目录。例如:$HOME/store/consumequeue/{topic}/{queueId}/。消费者正是通过这个文件来拉取消息的。
  • 文件结构: 每个 ConsumeQueue 文件包含 30 万个条目(Entry),每个条目大小固定为 20 字节。因此每个文件约为 5.72 MB。
  • 条目内容(20 字节的设计极其精妙):
    1. CommitLog Offset(8字节): 消息在 CommitLog 中的物理偏移量。
    2. Size(4字节): 消息的大小。
    3. 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):
    1. Header(40字节): 包含文件的起始/结束时间、最小/最大物理偏移量、哈希槽数量、索引项数量等元数据。
    2. Hash Slot(哈希槽,500万个): 每个槽 4 字节,存储该 Hash 对应的最新的 Index Item 的序号。
    3. Index Item(索引项,2000万个): 每个条目 20 字节。
  • Index Item 内容(20字节):
    1. Key Hash(4字节): Message Key 的哈希值。
    2. CommitLog Offset(8字节): 消息在 CommitLog 的物理偏移量。
    3. Timestamp(4字节): 消息存储时间与文件起始时间的差值(节省空间)。
    4. Next Index No(4字节): 发生 Hash 冲突时,指向前一个冲突项的序号(单向链表结构)。

4. 这三者是如何协同工作的?

了解了三者的结构,我们来看看一条消息从写入到被消费、被查询的全流程(这也是 RocketMQ 面试最常考的地方):

A. 消息写入(同步写,异步建索引)

  1. Producer 发送消息到 Broker。
  2. Broker 将消息顺序追加CommitLog 中。只要写入 CommitLog 成功,就向 Producer 返回成功。
  3. Broker 后台有一个名为 ReputMessageService 的异步线程,它会不断地读取 CommitLog 中新写入的数据。
  4. 该线程解析消息的 Topic、QueueId、Key 等信息,异步、实时地生成 ConsumeQueueIndexFile 的索引条目

B. 消息消费(基于 ConsumeQueue)

  1. Consumer 发起拉取请求,带上自己的 Topic、QueueId 以及逻辑偏移量(Offset)。
  2. Broker 找到对应的 ConsumeQueue 文件,定位到 20 字节的索引条目。
  3. 如果消费者配置了 Tag 过滤,Broker 首先对比条目中的 Tag HashCode,不匹配则跳过。
  4. 匹配成功后,取出 CommitLog OffsetSize
  5. 拿着 Offset 和 Size 去 CommitLog 中读取真正的消息主体,返回给 Consumer。

C. 消息查询(基于 IndexFile)

  1. 用户在控制台输入 Message Key(如订单号 Order123)。
  2. Broker 将 Order123 进行 Hash 计算。
  3. IndexFile 中找到对应的 Hash 槽。
  4. 顺着 Hash 槽的链表(Next Index No)遍历,比对时间戳等信息。
  5. 找到匹配项后,取出 CommitLog Offset,去 CommitLog 中读出完整消息返回。

总结

RocketMQ 的文件存储设计完美诠释了“空间换时间”“读写分离(物理写、逻辑读)”的理念:

  • CommitLog 保证了极高的写入性能。
  • ConsumeQueue 保证了高效的按 Topic 顺序消费。
  • IndexFile 提供了灵活的业务键查询能力。
    配合 Linux 的 PageCacheMmap(内存映射) 技术,使得这套基于磁盘的存储系统能够拥有接近内存级别的读写效率。