RocketMQ原理学习---Broker消息接收处理

谁践踏了优雅 2022-04-12 08:10 250阅读 0赞

上一篇博客[《RocketMQ原理学习---Producer消息发送》][RocketMQ_---Producer]中我们简单了解了RocketMQ生产者消息发送的过程，接下来我们看看Broker是如何处理接收到的消息。

RocketMQ的Broker接收消息涉及到很多操作，首先我们需要对RocketMQ所保存的消息文件目录及文件有所了解，RocketMQ消息数据保存目录及文件名如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70][]

介绍：

（1）commitlog：commitlog目录下为消息文件内容，所有的消息都会写到同一个文件中，文件大小为1G，超过1G会新创建一个文件，消息持久化有效时间为3天。

（2）consumequeue：consumequeue目录下是以Topic作为文件名称，每个Topic下又以queue id作为文件夹分组，用于记录每个消息在commitlog目录下文件中消息的位置，consumequeue目录下文件中记录的内容格式为 queueoffset/size/tag

（3）index：消息索引文件信息，根据topic和tag名称生成哈希键，值为消息在commitlog的偏移量。

（4）config：主要记录一些消费者的配置信息

consumerFilter：消费者的消息过滤器

consumeroffset：集群消息每个消费者的消费消息偏移量由name server来维护

delayoffset：延时消息的偏移量

topics：所有的topic的一些配置信息

subscriptionGroup：订阅分组信息

### **1、消息数据持久化** ###

Broker会将生产者发送过来的所有消息追加到commitlog目录消息持久化文件中，默认配置下，第一个commitLog的路径为~/store/commitlog/00000000000000000000 . 文件的大小默认为1G，消费者根据消费者队列的信息从持久化文件中随机读取消息。

Broker消息接收过程如下：

（1）Broker在启动过程中会在MappedFile初始化一个MappedByteBuffer对象，用这个对象将消息数据持久化到文件中。

（2）SendMessageProcessor的processRequest用来处理接收的消息，对于接收的消息经过转化之后调用DefaultMessageStore的putMessage方法

（3）在DefaultMessageStore的putMessage中会校验消息，并调用CommitLog的putMessage方法

（4）在CommitLog的putMessage会获取MappedFile对象，将消息追加到commitlog目录的文件中

（5）在MappedFile中生成byteBuffer对象调用put方法，将消息进行初始化操作。

Broker消息接收处理代码流程：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 1][]

Broker将所有的消息追加到commitlog目录下的文件中进行持久化，所以消费者在消费消息时需要从commitlog文件中读取消息信息，为了更快的读取消费者所关心的数据，RocketMQ会另外保存消费者消费读取文件的一些信息。

commitlog文件中一条数据内容包含以下相关属性：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 2][]

### **2、初始化consumequeue信息** ###

由于Broker将所有的消息都持久化到commitlog目录下的文件中，在消费消息时不可能将整个文件读取出来过滤想要消费的文件，因此Broker将每个Topic消息在commitlog目录下文件所在的位置信息写到如下路径中：

/store/consumequeue/${topicName}/${queueId}/${fileName}

这样Broker可以从consumequeue目录下的文件中读取每个消息的位置信息，就可以从commitlog目录下随机读取文件获取消息了。consumequeue相关信息写到consumequeue目录下文件代码执行流程如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 3][]

ConsumeQueue的结构如图：

[![E3_80_90RocketMQ_E6_BA_90_E7_A0_81_E5_AD_A6_E4_B9_A0_E3_80_916-_E6_B6_88_E6_81_AF_E5_AD_98_E5_82_A8_4.png][]][E3_80_90RocketMQ_E6_BA_90_E7_A0_81_E5_AD_A6_E4_B9_A0_E3_80_916-_E6_B6_88_E6_81_AF_E5_AD_98_E5_82_A8_4.png]

ConsumeQueue 文件的默认大小为30W 20，20是 queue中 每个存储单元的大小，每个存储单元对应 CommitLog 中的一个消息。即一个 ConsumeQueue 文件可以对应 30W 个消息。一个 CommitLog 的大小为 1G, 假设每个消息的大小为 1 K，则一个 CommitLog 可以存储 1024 1024 个消息，那就需要 1024 \* 1024 / 30W，向上取整4个ConsumeQueue. 所以，每个Topic可能有多个ConsumeQueue, 每个queue有自己的id, 对应的 fileName 为它的offset.

ConsumeQueue 中每个单元的大小为20字节，包含 commitLog offset/Size/Tag hashCode/ 三个字段，前面两个字段可以用于去 commitLog中查找真正的消息内容，最后一个可以用于在broker端根据tag过滤消息，当然，这个过滤不是完全靠谱的，不同的tag可能有相同的hashCode，Consumer端还可以根据自己的需要做进一步精确的过滤。

### 3、索引消息初始化 ###

Broker在初始化完consumequeue目录下的文件之后就是写index目录下的索引文件了，流程如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 4][]

说明：

（1）前三个流程都是触发动作的，不做过多介绍

（2）在第4步中调用buildIndex方法，首先会判断消息类型，如果是回滚的事物消息则不做处理直接返回，然后会根据消息的topic和uniqueKey生成哈希值

public void buildIndex(DispatchRequest req) {
            IndexFile indexFile = retryGetAndCreateIndexFile();
            if (indexFile != null) {
                long endPhyOffset = indexFile.getEndPhyOffset();
                DispatchRequest msg = req;
                String topic = msg.getTopic();
                String keys = msg.getKeys();
                if (msg.getCommitLogOffset() < endPhyOffset) {
                    return;
                }
    
                final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
                switch (tranType) {
                    case MessageSysFlag.TRANSACTION_NOT_TYPE:
                    case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
                    case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                        break;
                    case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                        return;
                }
    
                if (req.getUniqKey() != null) {
                    indexFile = putKey(indexFile, msg, buildKey(topic, req.getUniqKey()));
                    if (indexFile == null) {
                        log.error("putKey error commitlog {} uniqkey {}", req.getCommitLogOffset(), req.getUniqKey());
                        return;
                    }
                }
    
                if (keys != null && keys.length() > 0) {
                    String[] keyset = keys.split(MessageConst.KEY_SEPARATOR);
                    for (int i = 0; i < keyset.length; i++) {
                        String key = keyset[i];
                        if (key.length() > 0) {
                            indexFile = putKey(indexFile, msg, buildKey(topic, key));
                            if (indexFile == null) {
                                log.error("putKey error commitlog {} uniqkey {}", req.getCommitLogOffset(), req.getUniqKey());
                                return;
                            }
                        }
                    }
                }
            } else {
                log.error("build index error, stop building index");
            }
        }

（3）在第5步中会将哈希值，消息在commitlog文件中的偏移量等信息作为信息调用indexFile的putKey方法初始化到文件中。

private IndexFile putKey(IndexFile indexFile, DispatchRequest msg, String idxKey) {
            for (boolean ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp()); !ok; ) {
                log.warn("Index file [" + indexFile.getFileName() + "] is full, trying to create another one");
    
                indexFile = retryGetAndCreateIndexFile();
                if (null == indexFile) {
                    return null;
                }
    
                ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp());
            }
    
            return indexFile;
        }

消息格式：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 5][]

首先会去 IndexFileList 中获取一个最后一个 IndexFile, 判断它有没有被写满，如果满了就新建一个，否则就返回它。IndexFile的文件名是根据时间戳生成的。

如果消息的properties中设置了 UNIQ\_KEY 这个属性，就用 topic + “\#” + UNIQ\_KEY的value 作为 key 来做put操作。如果消息设置了 KEYS 属性（多个KEY以空格分隔），也会用 topic + “\#” + KEY 来做索引。

1.  计算槽位，slotPos = hash(key) % hashSlotNum, hashSlotNum 默认是500W.
2.  计算当前这个槽位的绝对地址（或者说在当前这个mappedFile中的内存地址） absSlotPos = IndexHeader.INDEX\_HEADER\_SIZE + slotPos \* hashSlotSize. headerSize 是 40个字节，按顺序依次保存这几个信息：
    
     *  Long beginTimestamp(8 Byte)
     *  Long endTimestamp(8 Byte)
     *  Long beginPhyoffset(8 Byte)
     *  Long endPhyoffset(8 Byte)
     *  Integer hashSlotcount(4 Byte)，看了代码，好像和indexCount的数量是一致的，不知道有啥用
     *  Integer indexCount(4 Byte), 索引的个数
3.  读取当前这个槽位的值 slotValue，如果之前之前这个槽位没有被占用过(Hash冲突的情况)，这个值会是0.
4.  计算真正写索引数据的地址, absIndexPos = IndexHeader.INDEX\_HEADER\_SIZE + this.hashSlotNum *hashSlotSize + this.indexHeader.getIndexCount() *indexSize;从absIndexPos这个位置开始,写入索引数据。

索引数据包括 Key Hash/CommitLog Offset/Timestamp/NextIndex offset 这四个字段，一共20 byte. NextIndex offset 即前面读出来的 slotValue, 如果有 hash冲突，就可以用这个字段将所有冲突的索引用链表的方式串起来了。Timestamp 记录的是消息storeTimestamp之间的差，并不是一个绝对的时间。整个Index File的结构如图，40 Byte 的Header用于保存一些总的统计信息，4 *500W的 Slot Table并不保存真正的索引数据，而是保存每个槽位对应的单向链表的头。20 *2000W 是真正的索引数据。即一个 Index File 可以保存 2000W个索引。

参考文章：

[https://blog.csdn.net/mr253727942/article/details/55805876][https_blog.csdn.net_mr253727942_article_details_55805876]

[【RocketMQ源码学习】6-消息存储][RocketMQ_6-]

[RocketMQ_---Producer]: https://blog.csdn.net/qq924862077/article/details/83928643
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70]: /images/20220412/f998668b749e4388af726b1b22e014c2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 1]: /images/20220412/72d29ee74d8c4af588a0f3d37f4a339c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 2]: /images/20220412/2062cc387461460991901298cef3b030.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 3]: /images/20220412/7cca8d352d6d4edfb9fd5e59f46d5bd8.png
[E3_80_90RocketMQ_E6_BA_90_E7_A0_81_E5_AD_A6_E4_B9_A0_E3_80_916-_E6_B6_88_E6_81_AF_E5_AD_98_E5_82_A8_4.png]: /images/20220412/0dfd7e265a1c4c8aa795ee6ff7fef2c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 4]: /images/20220412/9cc88ed6a7154c9888c6051c2ad0c6d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 5]: /images/20220412/bcd4ff2d3cc845aaa3b46666f7684b9c.png
[https_blog.csdn.net_mr253727942_article_details_55805876]: https://blog.csdn.net/mr253727942/article/details/55805876
[RocketMQ_6-]: https://fdx321.github.io/2017/08/22/%E3%80%90RocketMQ%E6%BA%90%E7%A0%81%E5%AD%A6%E4%B9%A0%E3%80%916-%E6%B6%88%E6%81%AF%E5%AD%98%E5%82%A8/