RocketMQ——消息的存储

淩亂°似流年 2021-09-26 23:00 390阅读 0赞

RocketMQ的消息存储是由consume queue和commit log配合完成的。

### **1、Consume Queue** ###

consume queue是消息的逻辑队列，相当于字典的目录，用来指定消息在物理文件commit log上的位置。

我们可以在配置中指定consumequeue与commitlog存储的目录

每个topic下的每个queue都有一个对应的consumequeue文件，比如：

$\{rocketmq.home\}/store/consumequeue/$\{topicName\}/$\{queueId\}/$\{fileName\}

Consume Queue文件组织，如图所示：

![20201228180014725.png][]

Consume Queue文件组织示意图

1.根据topic和queueId来组织文件，图中TopicA有两个队列0,1，那么TopicA和QueueId=0组成一个ConsumeQueue，TopicA和QueueId=1组成另一个ConsumeQueue。

2.按照消费端的GroupName来分组重试队列，如果消费端消费失败，消息将被发往重试队列中，比如图中的%RETRY%ConsumerGroupA。

3.按照消费端的GroupName来分组死信队列，如果消费端消费失败，并重试指定次数后，仍然失败，则发往死信队列，比如图中的%DLQ%ConsumerGroupA。

死信队列（Dead Letter Queue）一般用于存放由于某种原因无法传递的消息，比如处理失败或者已经过期的消息。

Consume Queue中存储单元是一个20字节定长的二进制数据，顺序写顺序读，如下图所示：

![20201228180034542.png][]

consumequeue文件存储单元格式

1.  CommitLog Offset是指这条消息在Commit Log文件中的实际偏移量
2.  Size存储中消息的大小
3.  Message Tag HashCode存储消息的Tag的哈希值：主要用于订阅时消息过滤（订阅时如果指定了Tag，会根据HashCode来快速查找到订阅的消息）

### **2、Commit Log** ###

CommitLog：消息存放的物理文件，每台 broker上的 commitlog被本机所有的 queue共享，不做任何区分。

文件的默认位置如下，仍然可通过配置文件修改：

*$\{user.home\}*  *\\store\\$* *\{commitlog\}* *\\$* *\{fileName\}*

CommitLog的消息存储单元长度不固定，文件顺序写，随机读。消息的存储结构如下表所示，按照编号顺序以及编号对应的内容依次存储。

Commit Log存储单元结构图:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d3ZDExNTQ5NzgzNTI_size_16_color_FFFFFF_t_70][]

### **3、消息存储实现** ###

消息存储实现，比较复杂。

// Set the storage time
    msg.setStoreTimestamp(System.currentTimeMillis());
    // Set the message body BODY CRC (consider the most appropriate setting
    msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
    StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
    synchronized (this) {
        long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
        // Here settings are stored timestamp, in order to ensure an orderly global
        msg.setStoreTimestamp(beginLockTimestamp);
        // MapedFile：操作物理文件在内存中的映射以及将内存数据持久化到物理文件中
        MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile();
        // 将Message追加到文件commitlog
        result = mapedFile.appendMessage(msg, this.appendMessageCallback);
        switch (result.getStatus()) {
        case PUT_OK:break;
        case END_OF_FILE:
             // Create a new file, re-write the message
             mapedFile = this.mapedFileQueue.getLastMapedFile();
             result = mapedFile.appendMessage(msg, this.appendMessageCallback);
         break;
         DispatchRequest dispatchRequest = new DispatchRequest(
                    topic,// 1
                    queueId,// 2
                    result.getWroteOffset(),// 3
                    result.getWroteBytes(),// 4
                    tagsCode,// 5
                    msg.getStoreTimestamp(),// 6
                    result.getLogicsOffset(),// 7
                    msg.getKeys(),// 8
                    /**
                     * Transaction
                     */
                    msg.getSysFlag(),// 9
                    msg.getPreparedTransactionOffset());// 10
        // 1.分发消息位置到ConsumeQueue
        // 2.分发到IndexService建立索引
        this.defaultMessageStore.putDispatchRequest(dispatchRequest);
    }

### **4、消息的索引文件** ###

如果一个消息包含key值的话，会使用IndexFile存储消息索引，文件的内容结构如图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d3ZDExNTQ5NzgzNTI_size_16_color_FFFFFF_t_70 1][]

消息索引

索引文件主要用于根据key来查询消息的，流程主要是：

1.根据查询的 key 的 hashcode%slotNum 得到具体的槽的位置(slotNum 是一个索引文件里面包含的最大槽的数目，例如图中所示 slotNum=5000000)

2.根据 slotValue(slot 位置对应的值)查找到索引项列表的最后一项(倒序排列,slotValue 总是指向最新的一个索引项)

3.遍历索引项列表返回查询时间范围内的结果集(默认一次最大返回的 32 条记录)

[20201228180014725.png]: /images/20210923/a40fd05bacf34b4bb9a396ad18d60d58.png
[20201228180034542.png]: /images/20210923/bb968a034ae04afc9121eec981833493.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d3ZDExNTQ5NzgzNTI_size_16_color_FFFFFF_t_70]: /images/20210923/90301149381e447793e5b36a6228688a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d3ZDExNTQ5NzgzNTI_size_16_color_FFFFFF_t_70 1]: /images/20210923/cf5cde81be9b41dda5c4de17cdc684a9.png