【RocketMQ】RocketMq之IndexFile深入研究

一：RocketMq 整体文件存储介绍

• CommitLog：存储消息的元数据。所有消息都会顺序存⼊到CommitLog⽂件当中。CommitLog由多个⽂件组成，每个⽂件固定⼤⼩1G。以第⼀条消 息的偏移量为⽂件名。
• ConsumerQueue：存储消息在CommitLog的索引。⼀个MessageQueue⼀个⽂件，记录当前MessageQueue被哪些消费者组消费到了哪⼀条CommitLog。
• IndexFile：为了消息查询提供了⼀种通过key或时间区间来查询消息的⽅法，这种通过IndexFile来查找消息的⽅法不影响发送与消费消息的主流程。

这篇文章主要介绍IndexFile的研究，以rocketmq5.3.0版本作为研究。

二：IndexFile的文件结构

文件整理格式，如下图2-1所示

                                                图2-1 IndexFile 文件结构图

IndexFile 文件格式

• 文件名：以时间戳命名（例如 20240301120000000），表示该文件索引的消息的时间范围。

• 文件大小：默认为 400MB，可通过 maxIndexSize 配置调整。

• 存储路径：默认在 ~/store/index 目录下。

每个 IndexFile 文件由三部分组成：
1. 文件头部（Header）
2. 哈希槽（Hash Slot）区域
3. 索引条目（Index Entry）区域

1. 文件头部（Header）

2. 哈希槽（Hash Slot）区域

• 哈希槽数量：固定为 500 万个（5,000,000），每个哈希槽占 4 字节。

• 哈希函数：对消息的 Key（如 UNIQ_KEY 或 KEYS）进行哈希计算，得到槽位索引：
  slotPos = abs(hash(key)) % 5000000

每个哈希槽存储的是 索引条目区域 的起始位置（索引条目链表的头节点）。

3. 索引条目（Index Entry）区域

每个索引条目占 20 字节，包含以下字段：

 三：IndexFile 写入和查询流程

IndexFile 写入流程：

+---------------------+
| Producer 发送消息     |
+---------------------+
          |
          v
+---------------------+
| 提取消息的 Key        | --> 如 UNIQ_KEY 或 KEYS 属性
+---------------------+
          |
          v
+---------------------+
| 检查 IndexFile 容量   | --> 是否已满？(indexCount >= indexNum)
+---------------------+
          | 是
          v
+---------------------+
| 返回 false，写入失败   |
+---------------------+
          | 否
          v
+---------------------+
| 计算 Key 的哈希值     | --> `keyHash = indexKeyHashMethod(key)`
+---------------------+
          |
          v
+---------------------+
| 计算哈希槽位置         | --> `slotPos = keyHash % hashSlotNum`
+---------------------+
          |
          v
+---------------------+
| 计算哈希槽绝对位置      | --> `absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize`
+---------------------+
          |
          v
+---------------------+
| 读取哈希槽的当前值      | --> `slotValue = mappedByteBuffer.getInt(absSlotPos)`
+---------------------+
          |
          v
+---------------------+
| 校验 slotValue 有效性  | --> 是否无效？(slotValue <= invalidIndex || slotValue > indexCount)
+---------------------+
          | 是
          v
+---------------------+
| 将 slotValue 设为无效   | --> `slotValue = invalidIndex`
+---------------------+
          | 否
          v
+---------------------+
| 计算时间差 (timeDiff)  | --> `timeDiff = (storeTimestamp - beginTimestamp) / 1000`
+---------------------+
          |
          v
+---------------------+
| 处理 timeDiff 边界值   | --> 确保 `0 <= timeDiff <= Integer.MAX_VALUE`
+---------------------+
          |
          v
+---------------------+
| 计算索引条目绝对位置     | --> `absIndexPos = IndexHeader.INDEX_HEADER_SIZE + hashSlotNum * hashSlotSize + indexCount * indexSize`
+---------------------+
          |
          v
+---------------------+
| 写入索引条目内容        |
| - keyHash            |
| - phyOffset          |
| - timeDiff           |
| - slotValue (nextIndex)|
+---------------------+
          |
          v
+---------------------+
| 更新哈希槽指向新条目     | --> `mappedByteBuffer.putInt(absSlotPos, indexCount)`
+---------------------+
          |
          v
+---------------------+
| 更新 IndexFile 头部信息 |
| - 若 indexCount <= 1，更新 beginPhyOffset 和 beginTimestamp |
| - 若 slotValue 无效，增加 hashSlotCount |
| - 增加 indexCount      |
| - 更新 endPhyOffset 和 endTimestamp |
+---------------------+
          |
          v
+---------------------+
| 返回 true，写入成功     |
+---------------------+
          |
          v
+---------------------+
| IndexFile 是否已满？   | -- 是 --> 创建新 IndexFile
| (文件大小 ≥ 400MB)    |
+---------------------+  

源码入口：org.apache.rocketmq.store.index.IndexFile#putKey

IndexFile 查询流程：

+---------------------+
| Consumer 根据 Key 查询 |
+---------------------+
          |
          v
+---------------------+
| 计算 Key 的哈希值     | --> `keyHash = Math.abs(key.hashCode())`
+---------------------+
          |
          v
+---------------------+
| 计算哈希槽位置         | --> `slotPos = keyHash % 5,000,000`
+---------------------+
          |
          v
+---------------------+
| 读取哈希槽的链表头位置   | --> `slotValue = mappedByteBuffer.getInt(slotPos * 4)`
+---------------------+
          |
          v
+---------------------+  
| 遍历链表条目           |
| while (slotValue > 0)|
+---------------------+
          |
          v
+---------------------+
| 读取索引条目：         |
| - keyHashRead       |
| - phyOffset         |
| - timeDiff          |
| - nextIndex         |
+---------------------+
          |
          v
+---------------------+
| 检查时间范围是否匹配？   | --> `storeTime = beginTimestamp + timeDiff * 1000`
| (storeTime ∈ [begin, end]?)|
+---------------------+
          | 否
          |------------------> 跳过，继续下一个条目
          | 是
          v
+---------------------+
| 比对 keyHashRead 和 keyHash |
| (是否相等？)          |
+---------------------+
          | 否
          |------------------> 跳过，继续下一个条目
          | 是
          v
+---------------------+
| 从 CommitLog 读取实际 Key |
| (检查 Key 是否一致？)    |
+---------------------+
          | 否
          |------------------> 跳过，继续下一个条目
          | 是
          v
+---------------------+
| 返回 phyOffset       | --> 添加到结果列表
+---------------------+
          |
          v
+---------------------+
| slotValue = nextIndex| --> 继续遍历下一个条目
+---------------------+
          |
          v
+---------------------+
| 遍历结束，返回结果列表   |
+---------------------+

源码入口：org.apache.rocketmq.store.index.IndexService#queryOffset

四：IndexFile解决hash冲突问题思想

RocketMQ 的 IndexFile 通过 链地址法（Chaining） 解决哈希冲突问题，其核心思想是将哈希到同一槽位的多个索引条目组织成链表结构，并通过哈希槽（Hash Slot）与索引条目（Index Entry）的关联实现高效写入和查询。以下是具体实现思想及关键设计：

1. 哈希冲突的背景

• 哈希冲突：不同 Key 经过哈希函数计算后可能得到相同的哈希值，导致被分配到同一个哈希槽。

• 问题：若不处理冲突，后续 Key 的索引会覆盖已有数据，导致查询结果错误。

2. 解决冲突的核心思想：链地址法

RocketMQ 的 IndexFile 采用 单链表 结构管理同一哈希槽下的所有冲突条目，具体流程如下：

(1) 写入时的链表插入

• 新条目插入链表头部：
  当新 Key 的哈希值与某槽位已有条目冲突时，新条目会被插入链表头部，并更新哈希槽指针指向新条目。

  // 新条目的 nextIndex 指向原头节点
  this.mappedByteBuffer.putInt(absIndexPos + 16, slotValue);
  // 更新哈希槽指针为新条目位置
  this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());

   

  • 优势：插入时间复杂度为 O(1)，无需遍历链表。

(2) 查询时的链表遍历

• 遍历链表比对 Key：
  查询时，从哈希槽指向的链表头节点开始，依次遍历所有条目，通过两次比对（哈希值 + 实际 Key）过滤冲突。

  while (nextIndexToRead > 0) {
      // 1. 读取条目内容
      int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);
      long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);
      // 2. 比对哈希值
      if (keyHashRead == keyHash) {
          // 3. 从 CommitLog 读取实际 Key 比对
          String keyStored = readKeyFromCommitLog(phyOffsetRead);
          if (key.equals(keyStored)) {
              phyOffsets.add(phyOffsetRead);
          }
      }
      // 4. 移动到下一个节点
      nextIndexToRead = prevIndexRead;
  }

3. 关键设计优化

(1) 哈希槽数量固定

• 默认 500 万个哈希槽：

  private static final int HASH_SLOT_NUM = 5000000; // 默认槽数

  • 目的：通过大量槽位减少哈希冲突的概率，使冲突链表尽可能短。

  • 权衡：槽数过多会占用更多内存，但查询效率更高。

(2) 时间范围过滤

• 索引条目存储时间差（timeDiff）：
  每个索引条目记录消息存储时间与 IndexFile 起始时间的差值（秒级），查询时快速过滤掉不满足时间范围的条目。

  long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff * 1000L;
  if (timeRead < begin || timeRead > end) {
      continue; // 跳过不符合时间条件的条目
  }

  • 优势：减少无效条目的遍历，提升查询性能。

(3) 文件滚动（Rolling）

• 按时间或大小滚动：
  IndexFile 文件默认大小上限为 400MB，或时间跨度超过阈值时，创建新文件。

  • 目的：避免单个文件过大导致链表过长，同时支持按时间范围快速定位文件。

4. 示例场景

写入冲突场景

• Key1: Ea#20231001123456 → 哈希值 19583063 → 槽位 18332292

• Key2: FB#20231001123456 → 哈希值 19583063 → 槽位 18332292（冲突）

• 处理流程：

  1. Key1 写入槽位 18332292，链表头指向 Key1。

  2. Key2 写入时，插入链表头部，槽位指针更新为 Key2，Key2 的 nextIndex 指向 Key1。

查询冲突场景

• 查询 Key: Ea#20231001123456

  1. 哈希计算定位到槽位 18332292。

  2. 遍历链表：

     • 先读取 Key2（哈希值匹配但 Key 不匹配，跳过）。

     • 再读取 Key1（哈希值 + Key 均匹配，返回 phyOffset）。

hash冲突代码调试示例

 public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("producerGroup");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();

        Message msg = new Message("Ea", "TagA" , ("消息1").getBytes(RemotingHelper.DEFAULT_CHARSET));
        msg.setKeys("20231001123456");
        producer.sendOneway(msg);

        Message msg2 = new Message("FB", "TagA" , ("消息3").getBytes(RemotingHelper.DEFAULT_CHARSET));
        msg2.setKeys("20231001123456");
        producer.sendOneway(msg2);
        
        producer.shutdown();
    }