RocketMQ 消息存储文件结构分析

RocketMQ 消息存储概述

RocketMQ 的消息存储是其核心功能之一，高效稳定的消息存储机制确保了消息的可靠持久化与快速读写。RocketMQ 采用了基于文件系统的存储方式，通过顺序写和随机读的策略来提升性能。这种存储策略利用了机械硬盘顺序写速度远高于随机写的特性，极大地提高了消息写入的效率。同时，在读取消息时，虽然是随机读，但通过合理的文件结构和索引设计，也能保证较高的读取性能。

RocketMQ 消息存储文件类型

CommitLog 文件

CommitLog 是 RocketMQ 存储消息的核心文件，所有的消息都顺序存储在这个文件中。这种设计保证了消息写入的高效性，因为顺序写避免了磁盘 I/O 寻道时间，从而显著提高了写入速度。每个 CommitLog 文件大小固定，默认是 1G。当一个 CommitLog 文件写满后，会创建一个新的 CommitLog 文件继续写入。

CommitLog 文件的命名规则为：以起始偏移量命名，例如 00000000000000000000，表示该文件的起始偏移量为 0。偏移量是消息在 CommitLog 文件中的物理位置，对于定位消息至关重要。

ConsumeQueue 文件

ConsumeQueue 是消息消费队列文件，它主要为消息消费提供索引服务。每个主题（Topic）的每个队列（Queue）都有对应的 ConsumeQueue 文件。ConsumeQueue 文件存储了指向 CommitLog 文件中消息的索引信息，包括消息在 CommitLog 中的偏移量、消息长度和消息 Tag 的哈希值等。

ConsumeQueue 文件的结构设计有助于快速定位消息，消费者在拉取消息时，通过 ConsumeQueue 文件可以快速找到对应的 CommitLog 位置，从而读取消息内容。这种设计解耦了消息存储和消息消费，使得消息消费的并发性能得到提升。

IndexFile 文件

IndexFile 文件用于为消息提供基于 Key 或者时间的索引。当消息发送时，如果设置了 Key，RocketMQ 会为该消息在 IndexFile 中创建索引。IndexFile 文件由多个槽位（Slot）组成，每个槽位指向一个具体的索引项。索引项记录了消息在 CommitLog 中的偏移量、消息 Key 的哈希值等信息。

通过 IndexFile，用户可以根据消息 Key 快速定位到消息在 CommitLog 中的位置，方便进行消息查询和回溯等操作。这在一些需要根据特定条件查询消息的场景中非常有用。

CommitLog 文件结构详解

消息格式

CommitLog 文件中的每条消息都有固定的格式，其结构如下：

1. 魔数（4 字节）：用于标识文件格式版本，固定值为 0xdaa320a7。
2. 长度（4 字节）：表示整个消息体的长度，包括消息内容、属性等。
3. CRC 校验码（4 字节）：用于校验消息的完整性。
4. 存储时间戳（8 字节）：消息存储到 CommitLog 的时间。
5. 队列 ID（4 字节）：消息所属的队列 ID。
6. 标记（4 字节）：一些额外的标记信息。
7. 消息体长度（4 字节）：消息内容的长度。
8. 消息体（N 字节）：实际的消息内容。
9. 属性长度（4 字节）：消息属性的长度。
10. 属性（N 字节）：消息的属性，以 Key - Value 形式存储。

以下是使用 Java 代码模拟构建 CommitLog 消息格式的示例：

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

public class CommitLogMessageBuilder {
    private static final int MAGIC_NUMBER = 0xdaa320a7;
    private static final int CRC_LENGTH = 4;
    private static final int TIMESTAMP_LENGTH = 8;
    private static final int QUEUE_ID_LENGTH = 4;
    private static final int FLAG_LENGTH = 4;
    private static final int BODY_LENGTH_LENGTH = 4;
    private static final int PROPERTY_LENGTH_LENGTH = 4;

    public static byte[] buildMessage(String messageBody, Map<String, String> properties) {
        byte[] bodyBytes = messageBody.getBytes();
        byte[] propertyBytes = serializeProperties(properties);

        int totalLength = 4 + 4 + CRC_LENGTH + TIMESTAMP_LENGTH + QUEUE_ID_LENGTH + FLAG_LENGTH + BODY_LENGTH_LENGTH + bodyBytes.length + PROPERTY_LENGTH_LENGTH + propertyBytes.length;
        ByteBuffer byteBuffer = ByteBuffer.allocate(totalLength);

        byteBuffer.putInt(MAGIC_NUMBER);
        byteBuffer.putInt(totalLength);
        // 这里假设 CRC 校验码为 0，实际需要计算
        byteBuffer.putInt(0);
        byteBuffer.putLong(System.currentTimeMillis());
        // 假设队列 ID 为 0
        byteBuffer.putInt(0);
        // 假设标记为 0
        byteBuffer.putInt(0);
        byteBuffer.putInt(bodyBytes.length);
        byteBuffer.put(bodyBytes);
        byteBuffer.putInt(propertyBytes.length);
        byteBuffer.put(propertyBytes);

        return byteBuffer.array();
    }

    private static byte[] serializeProperties(Map<String, String> properties) {
        if (properties == null || properties.isEmpty()) {
            return new byte[0];
        }
        StringBuilder stringBuilder = new StringBuilder();
        for (Map.Entry<String, String> entry : properties.entrySet()) {
            stringBuilder.append(entry.getKey()).append('=').append(entry.getValue()).append(';');
        }
        if (stringBuilder.length() > 0) {
            stringBuilder.setLength(stringBuilder.length() - 1);
        }
        return stringBuilder.toString().getBytes();
    }

    public static void main(String[] args) {
        String messageBody = "Hello, RocketMQ!";
        Map<String, String> properties = new HashMap<>();
        properties.put("key1", "value1");
        byte[] message = buildMessage(messageBody, properties);
        System.out.println("Message length: " + message.length);
    }
}

写入流程

1. 当生产者发送消息时，RocketMQ 首先将消息写入到内存中的 PageCache（页缓存）。PageCache 是操作系统内核提供的一种缓存机制，它可以提高文件 I/O 的性能。
2. 然后，通过异步刷盘机制将 PageCache 中的数据持久化到 CommitLog 文件中。刷盘策略分为同步刷盘和异步刷盘两种：
   • 同步刷盘：消息写入 PageCache 后，会等待刷盘操作完成，确保消息已经持久化到磁盘，再返回成功响应给生产者。这种策略保证了消息的可靠性，但会降低写入性能。
   • 异步刷盘：消息写入 PageCache 后，立即返回成功响应给生产者，后台线程会定期将 PageCache 中的数据刷盘。这种策略提高了写入性能，但在系统崩溃等极端情况下，可能会丢失部分未刷盘的消息。

ConsumeQueue 文件结构详解

索引格式

ConsumeQueue 文件中的每个索引项占用固定的 20 字节，其结构如下：

1. 消息在 CommitLog 中的偏移量（8 字节）：用于定位消息在 CommitLog 文件中的物理位置。
2. 消息长度（4 字节）：消息的总长度，包括消息内容、属性等。
3. 消息 Tag 的哈希值（8 字节）：用于快速过滤消息。

以下是使用 Java 代码模拟构建 ConsumeQueue 索引项的示例：

import java.nio.ByteBuffer;

public class ConsumeQueueIndexBuilder {
    public static byte[] buildIndex(long commitLogOffset, int messageLength, long tagHash) {
        ByteBuffer byteBuffer = ByteBuffer.allocate(20);
        byteBuffer.putLong(commitLogOffset);
        byteBuffer.putInt(messageLength);
        byteBuffer.putLong(tagHash);
        return byteBuffer.array();
    }

    public static void main(String[] args) {
        long commitLogOffset = 1000L;
        int messageLength = 100;
        long tagHash = 123456789L;
        byte[] index = buildIndex(commitLogOffset, messageLength, tagHash);
        System.out.println("Index length: " + index.length);
    }
}

读取流程

1. 消费者在拉取消息时，首先从 ConsumeQueue 文件中读取索引项。根据消费者的消费进度（偏移量），定位到 ConsumeQueue 文件中的具体位置，读取相应的索引项。
2. 从索引项中获取消息在 CommitLog 中的偏移量和消息长度等信息。
3. 根据 CommitLog 偏移量，到 CommitLog 文件中读取实际的消息内容。

IndexFile 文件结构详解

索引项结构

IndexFile 文件由多个槽位（Slot）和索引项组成。每个槽位占用 4 字节，用于存储索引项的位置偏移。索引项的结构如下：

1. 消息在 CommitLog 中的偏移量（8 字节）：定位消息在 CommitLog 文件中的物理位置。
2. 消息 Key 的哈希值（8 字节）：用于快速查找消息。
3. 下一个相同哈希值的索引项偏移量（4 字节）：如果有多个消息的 Key 哈希值相同，通过这个偏移量可以找到下一个相同哈希值的索引项。
4. 时间戳（4 字节）：消息存储到 CommitLog 的时间。

以下是使用 Java 代码模拟构建 IndexFile 索引项的示例：

import java.nio.ByteBuffer;

public class IndexFileIndexBuilder {
    public static byte[] buildIndex(long commitLogOffset, long keyHash, int nextIndexOffset, int timestamp) {
        ByteBuffer byteBuffer = ByteBuffer.allocate(24);
        byteBuffer.putLong(commitLogOffset);
        byteBuffer.putLong(keyHash);
        byteBuffer.putInt(nextIndexOffset);
        byteBuffer.putInt(timestamp);
        return byteBuffer.array();
    }

    public static void main(String[] args) {
        long commitLogOffset = 2000L;
        long keyHash = 987654321L;
        int nextIndexOffset = 10;
        int timestamp = (int) (System.currentTimeMillis() / 1000);
        byte[] index = buildIndex(commitLogOffset, keyHash, nextIndexOffset, timestamp);
        System.out.println("Index length: " + index.length);
    }
}

查找流程

1. 当根据消息 Key 查找消息时，首先计算 Key 的哈希值。
2. 通过哈希值找到 IndexFile 中对应的槽位，槽位中存储了第一个具有相同哈希值的索引项的偏移量。
3. 根据偏移量读取索引项，比较索引项中的 Key 哈希值是否与计算得到的哈希值一致。如果一致，则找到对应的消息；如果不一致，则根据下一个相同哈希值的索引项偏移量继续查找。

RocketMQ 存储文件管理

文件命名与路径

RocketMQ 的存储文件都存储在 broker 的存储目录下，默认路径为 $ROCKET_HOME/store。CommitLog 文件位于 commitlog 子目录下，ConsumeQueue 文件位于 consumequeue 子目录下，IndexFile 文件位于 index 子目录下。

文件命名规则如前文所述，CommitLog 文件以起始偏移量命名，ConsumeQueue 文件的命名格式为 topic/queueId/offset，IndexFile 文件的命名格式为 index_{timestamp}，其中 timestamp 为文件创建的时间戳。

文件清理机制

1. CommitLog 文件清理：当 CommitLog 文件中的消息都已经被消费，并且超过了保留时间（默认 72 小时），该文件会被标记为可删除。RocketMQ 会定期扫描 CommitLog 文件，删除这些过期的文件。
2. ConsumeQueue 文件清理：ConsumeQueue 文件的清理与 CommitLog 文件相关联。当对应的 CommitLog 文件被删除时，其关联的 ConsumeQueue 文件也会被删除。
3. IndexFile 文件清理：IndexFile 文件的清理同样依赖于消息的过期时间。当 IndexFile 中索引的消息都已经过期，该 IndexFile 文件会被删除。

性能优化与注意事项

性能优化

1. 合理配置刷盘策略：根据业务对消息可靠性和性能的要求，选择合适的刷盘策略。如果对消息可靠性要求极高，如金融业务，建议采用同步刷盘；如果对性能要求较高，对消息丢失有一定容忍度，可采用异步刷盘。
2. 优化 PageCache 配置：调整操作系统的 PageCache 相关参数，如 swappiness，可以减少内存交换，提高文件 I/O 性能。同时，合理分配系统内存，确保 PageCache 有足够的空间缓存数据。
3. 使用 SSD 硬盘：SSD 硬盘具有随机读写性能高的特点，使用 SSD 硬盘可以显著提升 RocketMQ 的读写性能，尤其是在随机读场景下。

注意事项

1. 文件系统选择：建议使用 ext4 或 XFS 等高性能文件系统，避免使用 FAT32 等不适合高性能 I/O 的文件系统。
2. 磁盘空间监控：定期监控 RocketMQ 存储目录的磁盘空间，避免因磁盘空间不足导致消息写入失败。可以设置磁盘空间预警机制，当磁盘空间低于一定阈值时，及时通知运维人员处理。
3. 索引文件维护：由于 IndexFile 文件主要用于消息查询，在高并发写入场景下，可能会导致索引文件增长过快。需要定期清理过期的索引文件，以避免占用过多磁盘空间。

通过深入了解 RocketMQ 的消息存储文件结构，我们可以更好地优化和管理 RocketMQ 集群，确保其在高并发、大规模消息处理场景下的性能和可靠性。无论是消息的写入、存储还是读取，每个环节的优化都对于整个消息系统的稳定性和高效性至关重要。同时，合理的文件管理和性能优化策略能够帮助我们充分发挥 RocketMQ 的优势，满足不同业务场景的需求。