RocketMQ：深入理解消息存储流程

蔚落 2023-10-02 13:50 21阅读 0赞

## 一、前言 ##

前面我们已经介绍过了消息存储在磁盘中的表现形式：[RocketMQ：消息整体存储架构（CommitLog、ConsumeQueue）][RocketMQ_CommitLog_ConsumeQueue]，Broker端如何接收Producer生产消息请求：[RocketMQ：深入理解Broker如何接收Producer生产消息请求？][RocketMQ_Broker_Producer]  
我们来继续分析RocketMQ的Broker端，一个生产消息请求到Broker后，Broker是如何保存这条消息的？

## 二、存储架构 ##

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16]  
每个Broker都对应有一个MessageStore服务，专门用来持久化Producer生产的消息。  
`MessageStore` 中持有一个`CommitLog`对象；CommitLog中维护了一个 `MappedFileQueue`，MappedFileQueue 中维护了多个 `MappedFile`，每个MappedFile都会映射到文件系统中一个文件，这些文件才是真正的存储消息的地方，MappedFile的文件名为它存储的第一条消息在CommitLog中的`全局物理偏移量`。

## 三、存储流程源码分析 ##

以下所有分析相关的源码注释请见GitHub中的release-4.8.0分支：[https://github.com/Saint9768/rocketmq/tree/rocketmq-all-4.8.0][https_github.com_Saint9768_rocketmq_tree_rocketmq-all-4.8.0]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_13_color_FFFFFF_t_70_g_se_x_16]

### 1、接收消息 ###

Broker端如何接收Producer的请求，我们在之前介绍过：[RocketMQ：深入理解Broker如何接收Producer生产消息请求？][RocketMQ_Broker_Producer]

我们现在接着SendMessageProcessor继续看消息是如何持久化的？  
在`SendMessageProcessor#asyncSendMessage()`方法调用MessageStore服务进行持久化消息操作。  
![在这里插入图片描述][14fa30af54374fba82a6f25f0c2fb253.png]  
接着看MessageStore\#`asyncPutMessage()`方法，其内部将调用`putMessage()`方法。  
![在这里插入图片描述][8f460168e3934b71900cbc2686397698.png]  
由于`MessageStore`是一个接口，我们看其实现`DefaultMessageStore`\#`putMessage()`方法。  
![在这里插入图片描述][5bf6f55909ea4d2aab42571e7351257a.png]  
OK了，入口就是这了：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 1]

### 2、存储消息到CommitLog ###

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 2]

#### 1. DefaultMessageStore\#putMessage()方法 ####

在上面我们已经介绍过了，Broker端接收完生产消息请求之后会进入到`DefaultMessageStore#putMessage()`方法。来我们看一下这个putMessage()都做了什么？我们这里将其分为四步进行介绍；  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 3]  
**第一步：校验消息存储服务Broker的状态；**  
包括：DefaultMessageStore消息存储服务状态不能为shutdown，broker的角色必须为mater，消息存储服务是可写的，操作系统页缓存不能是繁忙状态。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 4]  
**第二步：校验该批消息的topic和消息属性信息；**  
校验信息包括：topic的长度不能大于127B，消息属性的长度不能大于32767B。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 5]  
**第三步：调用CommitLog的putMessage()方法将消息存储到CommitLog中；**

我们先贴出`CommitLog#putMessage()`的源码，然后再一点一点的分析：

public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
        
        // Set the storage time
        // 设置存储时间
        msg.setStoreTimestamp(System.currentTimeMillis());
        // Set the message body BODY CRC (consider the most appropriate setting
        // on the client)
        // 设置消息体 CRC 校验
        msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
        // Back to Results
        AppendMessageResult result = null;
    
        StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
    
        String topic = msg.getTopic();
        int queueId = msg.getQueueId();
    
        // 设置消息事务类型
        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
        // TRANSACTION_NOT_TYPE 为普通消息、TRANSACTION_COMMIT_TYPE 为事务消息提交类型
        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
                || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
        
            // Delay Delivery
            // 获取消息延迟发送级别
            if (msg.getDelayTimeLevel() > 0) {
        
                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
        
                    msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                }
    
                // 设置延时发送的Topic
                topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
                // 根据延时发送级别确定队列ID
                queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
    
                // Backup real topic, queueId
                // 备份真正的Topic和队列ID
                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
                msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
    
                msg.setTopic(topic);
                msg.setQueueId(queueId);
            }
        }
    
        InetSocketAddress bornSocketAddress = (InetSocketAddress) msg.getBornHost();
        if (bornSocketAddress.getAddress() instanceof Inet6Address) {
        
            msg.setBornHostV6Flag();
        }
    
        InetSocketAddress storeSocketAddress = (InetSocketAddress) msg.getStoreHost();
        if (storeSocketAddress.getAddress() instanceof Inet6Address) {
        
            msg.setStoreHostAddressV6Flag();
        }
    
        long elapsedTimeInLock = 0;
    
        MappedFile unlockMappedFile = null;
        // 获取CommitLog的最后一个MappedFile
        MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
    
        // 写消息前加锁，写消息是串行的。
        putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
        try {
        
            long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
            this.beginTimeInLock = beginLockTimestamp;
    
            // Here settings are stored timestamp, in order to ensure an orderly
            // global
            // 设置存储时间戳，保证全局有序性
            msg.setStoreTimestamp(beginLockTimestamp);
    
            // 如果文件不存在，或者文件已满，创建新的MappedFile文件，并添加到mappedFileQueue中
            if (null == mappedFile || mappedFile.isFull()) {
        
                mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
            }
            if (null == mappedFile) {
        
                log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                beginTimeInLock = 0;
                return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
            }
    
            // 追加消息
            result = mappedFile.appendMessage(msg, this.appendMessageCallback);
            switch (result.getStatus()) {
        
                case PUT_OK:
                    break;
                // 如果消息已经写到了文件的结尾还没写完，则创建新的文件重新写入
                case END_OF_FILE:
                    unlockMappedFile = mappedFile;
                    // Create a new file, re-write the message
                    mappedFile = this.mappedFileQueue.getLastMappedFile(0);
                    if (null == mappedFile) {
        
                        // XXX: warn and notify me
                        log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                        beginTimeInLock = 0;
                        return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result);
                    }
                    result = mappedFile.appendMessage(msg, this.appendMessageCallback);
                    break;
                // 下面为存放失败的结果返回
                case MESSAGE_SIZE_EXCEEDED:
                case PROPERTIES_SIZE_EXCEEDED:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result);
                case UNKNOWN_ERROR:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
                default:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
            }
            // 计算存放消息消耗的时间
            elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
            beginTimeInLock = 0;
        } finally {
        
            putMessageLock.unlock();
        }
    
        if (elapsedTimeInLock > 500) {
        
            log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);
        }
    
        if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
        
            this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
        }
    
        PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
    
        // Statistics
        // 统计存放消息的数量和总大小
        storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
        storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
    
        // 处理磁盘刷新
        handleDiskFlush(result, putMessageResult, msg);
        // 处理高可用
        handleHA(result, putMessageResult, msg);
    
        return putMessageResult;
    }

由于`CommitLog.putMessage()`操作较为复杂，这里我们分5步进行分析；

##### 1）第一步：消息属性填充。 #####

给消息设置一些属性，包括：存储时间、消息体CRC校验；  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 6]

##### 2）第二步：延时消息相关设置。 #####

根据事务类型和消息延时发送级别决定是否–设置延时发送的Topic、queueId 并 备份真正的topic和queueID。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 7]  
只有当消息的事务类型为：普通消息、已提交的事务消息时，才会进一步判断消息是否设置了延时级别`delayTimeLevel`，如果设置了才会走到这个第二步。

##### 3）第三步：获取最后一个MappedFile，并往其中串行追加消息体。 #####

消息存储的核心步骤就在这里，而且这一步的细节点特别多。我们再分几步进行分析；  
**（1）首先获取CommitLog中的最后一个MappedFile；**  
![在这里插入图片描述][18f9d86eea0a4d7597cca94fc050fd60.png]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 8]  
**（2）在真正开始写消息之前要对CommitLog加锁，以保证消息串行化写入到CommitLog中；**  
![在这里插入图片描述][3f7236c8a8604a36a6b16d3285fcf397.png]  
既然要加锁，那加的是什么锁呢？

> 我们发现`PutMessageLock`是一个interface接口，其有两个实现：`PutMessageReentrantLock`、`PutMessageSpinLock`。它们一个采用ReentrantLock实现，一个采用CAS机制实现。  
> ![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 9]  
> ![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 10]

> 那么RocketMQ默认是使用的哪个锁？为什么要使用它？
> 
>  *  在`CommitLog进行实例化`时，会根据`MessageStoreConfig`配置类中的`useReentrantLockWhenPutMessage`属性来决定使用哪个锁？
>  *  因为`useReentrantLockWhenPutMessage`的`默认值是FALSE`，所以使用的是`PutMessageSpinLock`。
>  *  我们可以看到在`PutMessageSpinLock`中有说明，在`低锁竞争时使用`它。  
>     ![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 11]  
>     ![在这里插入图片描述][8746899cd8f44a29a3428eae1402e1d3.png]  
>     ![在这里插入图片描述][db3dd8357f094de88a6638b586233132.png]

**（3）设置存储时间戳；对获取到的MappedFile进行校验，确保有一个可用的MappedFile。**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 12]  
如果文件不存在 或者 已满，会创建一个新的MappedFile文件，并将其添加到`mappedFileQueue`中。

当需要新建一个MappedFile时，我们来看一下`mappedFileQueue.getLastMappedFile(0)`做了什么？  
![在这里插入图片描述][3ec62a6fa52e4ff9835cbff95fdd97f5.png]  
它又调用了MappedFileQueue中的一个同名方法，go on, go on，我们继续：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 13]  
通过获取MappedFiles集合中的`最后一个MappedFile的文件名`（起始offset） 和`文件中的消息数量mappedFileSize`，拼接出`将要创建的MappedFile` 和 下一个要创建的MappedFile的 `文件全路径名`；  
然后将他们作为一个请求放入到`AllocateMappedFileService`服务的`requestQueue`中，由AllocateMappedFileService服务`循环进行MMAP操作`，`分配MappedFile`。关于AllocateMappedFileService服务的MMAP操作，我们后面纤细介绍。

**（4）向MappedFile中追加消息，追加完消息之后释放锁。**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 14]  
直接向上面获取到的MappedFile追加消息，然后根据返回的状态进一步操作。

>  *  如果消息已经写到了`MappedFile文件的结尾还没写完`，那么就`创建`一个新的`MappedFile`文件 `重新写入消息`；
>  *  如果是存放消息失败，直接返回返回错误信息。
>  *  追加消息成功之后，统计消息存放消耗的时间。
>  *  最后把上面加的`锁给释放`了。

我们进入`MappedFile#appendMessage()`方法看一下消息是怎么追加到MappedFile文件的。  
![在这里插入图片描述][b54d3ac5fc234f35b2b0109efd0eff61.png]  
追加消息调用的是`MappedFile.appendMessage()`方法，此方法最终调用到`MappedFile.appendMessagesInner()`中：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 15]  
我们来分析下`appendMessagesInner()`中追加消息的的逻辑：

1.  获取当前MappedFile文件的写入位置，如果当前文件未写满，则进入追加逻辑；
2.  获取`MappedFile`中的 `writeBuffer`，如果 `writeBuffer 为空`，则`获取mappedByteBuffer`>

>  *  在MessageStore 初始化的时候，会初始化一个Buffer缓存池：`TransientStorePool`，TransientStorePool在初始化时会初始化若干DirectBuffer（默认5个），放入一个Deque中。MappedFile的writeBuffer就是从这个Deque中获取的。
>  *  而 `mappedByteBuffer` 类型为 `MappedByteBuffer`，`每个MappedFile都会映射到文件系统中的一个文件`，mappedByteBuffer 即为该文件在内存中的映射。
>  *  当追加消息到MappedFile中，会优先追加到 writeBuffer中。

1.  调用 `cb.doAppend()`追加消息;

> 几个参数：
> 
>  *  this.getFileFromOffset()：MappedFile的文件名（即MappedFile中第一个消息在CommitLog中的全局物理偏移量）。
>  *  byteBuffer：即MappedFile的内存缓冲区，也即是 2 中的writeBuffer或mappedByteBuffer。
>  *  this.fileSize - currentPos：fileSize为单个文件的额定大小，默认为1GB，currentPos为当前文件中已经写到什么位置，两个相减即为当前文件剩余容量。
>  *  messageExt：内部封装好的消息。

**writeBuffer从何而来？**  
![在这里插入图片描述][91ec08f8af4c41f6b4811dcac7332543.png]  
从这里我们可以看到，writeBuffer是在MappedFile初始化的时候从`TransientStorePool`中获取的。那么，**TransientStorePool又是怎么来的？**  
往上追了一下，没找到，就很尴尬。我想，我想，既然和消息存储有关，会不会在DefaultMessageStore中，呀，真巧，真在呢。  
在实例化`DefaultMessageStore`时，会实例化并初始化`transientStorePool`。  
![在这里插入图片描述][c9f6c4c11978467891d203baa7934eb4.png]  
我们看一下`transientStorePool.init()`，发现它会实例化poolSize个ByteBuffer放入到双向队列`availableBuffers`中。

private final Deque<ByteBuffer> availableBuffers;

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_19_color_FFFFFF_t_70_g_se_x_16]  
`poolSize的值`是多少？  
![在这里插入图片描述][9214a4f02c234bd08f4a4845591ffce2.png]  
在其构造函数中我们发现poolSize的值取自`MessageStoreConfig的transientStorePoolSize变量`。  
![在这里插入图片描述][ed6e424f0c3945de803c28828487825a.png]  
进入`MessageStoreConfig`类中我们可以看到`transientStorePoolSize`的默认值为5，所以这个`ByteBuffer的池子大小为5`。

**cb是个啥？从哪来的？**

> 在CommitLog中调用`appendMessagesInner()`时，传入的 cb 为：`this.appendMessageCallback`，它的类型为 `DefaultAppendMessageCallback`，实现了`AppendMessageCallback`接口。  
> ![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 16]

**所以，我们进入`CommitLog`的内部类`DefaultAppendMessageCallback`中看一下`doAppend()`都做了什么？**

**我们这里再把`doAppend()`方法拆个几步来看一下：**

**>1. 计算消息存储的各个属性，比如：消息长度、消息在ConsumeQueue中的长度等；**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 17]  
**>2. 判断消息体大小是否超过4M，超过则返回错误信息；**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 18]  
**>3. 判断MappedFile中的可用空间是否可以容纳下要追加的消息，如果不可以，返回错误码END\_OF\_FILE。**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 19]如果返回错误码为END\_OF\_FILE，接着CommitLog\#puMessage()方法中创建一个新的MappedFile，再重新写这条消息。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 20]  
**>4. 序列化消息内容，存储到内存缓存区中；**

// Initialization of storage space 初始化存储空间
     this.resetByteBuffer(msgStoreItemMemory, msgLen);
     // 1 TOTALSIZE -- 四节点用来存储消息的总大小
     this.msgStoreItemMemory.putInt(msgLen);
     // 2 MAGICCODE -- 四字节用来存储文件结尾空的魔数(类似于String字符串最后留的/0)
     this.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);
     // 3 BODYCRC -- 4字节的消息体CRC校验和
     this.msgStoreItemMemory.putInt(msgInner.getBodyCRC());
     // 4 QUEUEID -- 4字节的队列ID
     this.msgStoreItemMemory.putInt(msgInner.getQueueId());
     // 5 FLAG -- 4字节的Flag
     this.msgStoreItemMemory.putInt(msgInner.getFlag());
     // 6 QUEUEOFFSET -- 8字节的队列偏移量
     this.msgStoreItemMemory.putLong(queueOffset);
     // 7 PHYSICALOFFSET -- 8字节的物理偏移量
     this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());
     // 8 SYSFLAG -- 4字节的系统标识
     this.msgStoreItemMemory.putInt(msgInner.getSysFlag());
     // 9 BORNTIMESTAMP -- 8字节的消息出生时间戳
     this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
     // 10 BORNHOST -- 消息出生 host 主机
     this.resetByteBuffer(bornHostHolder, bornHostLength);
     this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
     // 11 STORETIMESTAMP -- 消息存储 host 主机
     this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
     // 12 STOREHOSTADDRESS
     this.resetByteBuffer(storeHostHolder, storeHostLength);
     this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));
     // 13 RECONSUMETIMES -- 重试次数
     this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
     // 14 Prepared Transaction Offset -- 8字节的事务准备偏移量
     this.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());
     // 15 BODY -- 消息长度
     this.msgStoreItemMemory.putInt(bodyLength);
     if (bodyLength > 0)
         // 消息内容
         this.msgStoreItemMemory.put(msgInner.getBody());
     // 16 TOPIC -- Topic长度
     this.msgStoreItemMemory.put((byte) topicLength);
     // Topic内容
     this.msgStoreItemMemory.put(topicData);
     // 17 PROPERTIES properties 长度
     this.msgStoreItemMemory.putShort((short) propertiesLength);
     if (propertiesLength > 0)
         this.msgStoreItemMemory.put(propertiesData);
    
     final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
     // Write messages to the queue buffer -- 消息写入到缓存区
     byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);

**>5. 返回追加消息成功的结果。**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 21]

##### 4）第四步：处理磁盘刷新； #####

即CommitLog数据从ByteBuffer刷到磁盘。详细的我们后面专门介绍。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 22]  
默认异步刷盘。

##### 5）第五步：处理高可用； #####

即主从Broker消息同步。详细的我们后面专门介绍。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 23]  
默认异步进行主从复制。

**第四步：统计消息存储花费的时间；如果存储失败，则增加存储失败的次数；**  
![在这里插入图片描述][07cf18d779684fd381e7919273152547.png]

### 3、CommitLog刷盘 ###

在`DefaultMessageStore服务启动`的时候，会启动CommitLog刷盘服务`flushCommitLogService`来将消息从直接内存ByteBuffer中刷到磁盘中。

### 4、后续操作 ###

同步CommitLog中的消息到ConsumeQueue，  
同步CommitLog中的消息到IndexFile。

## 四、总结 ##

RocketMQ 会创建多个MappedFile用来存储消息，每个MappedFile大小固定1G，其有自己的内存缓冲区和对应的系统文件；所有的MappedFile由CommitLog中的MappedFileQueue统一维护。

[RocketMQ_CommitLog_ConsumeQueue]: https://blog.csdn.net/Saintmm/article/details/120959054
[RocketMQ_Broker_Producer]: https://blog.csdn.net/Saintmm/article/details/121116644
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/a67c920f2ae34ff49bfa275146fb4a39.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[https_github.com_Saint9768_rocketmq_tree_rocketmq-all-4.8.0]: https://github.com/Saint9768/rocketmq/tree/rocketmq-all-4.8.0
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_13_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/cecee6393cc140c6aa35b06f42ac523b.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_13,color_FFFFFF,t_70,g_se,x_16
[14fa30af54374fba82a6f25f0c2fb253.png]: https://img-blog.csdnimg.cn/14fa30af54374fba82a6f25f0c2fb253.png
[8f460168e3934b71900cbc2686397698.png]: https://img-blog.csdnimg.cn/8f460168e3934b71900cbc2686397698.png
[5bf6f55909ea4d2aab42571e7351257a.png]: https://img-blog.csdnimg.cn/5bf6f55909ea4d2aab42571e7351257a.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/320cb03a888e41cd876b1c9fe250f8ce.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 2]: https://img-blog.csdnimg.cn/05354142140b4caf8aee19c597257844.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 3]: https://img-blog.csdnimg.cn/3a3a9b7b703a46e282f00f0e83c2c3ad.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 4]: https://img-blog.csdnimg.cn/5eed0e3328254f68b037918ca5221443.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 5]: https://img-blog.csdnimg.cn/32a6729e64234ddaa4b297bb73f332af.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 6]: https://img-blog.csdnimg.cn/c52888f145b54e01a2efcb81ec83db10.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 7]: https://img-blog.csdnimg.cn/7bdfb6eb5a4142038c4144b8058e0151.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[18f9d86eea0a4d7597cca94fc050fd60.png]: https://img-blog.csdnimg.cn/18f9d86eea0a4d7597cca94fc050fd60.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 8]: https://img-blog.csdnimg.cn/75c4cce71d3248f8a9f56eb9d3418a4a.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[3f7236c8a8604a36a6b16d3285fcf397.png]: https://img-blog.csdnimg.cn/3f7236c8a8604a36a6b16d3285fcf397.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 9]: https://img-blog.csdnimg.cn/35155ee9f464406a930ab4dfbc99f684.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 10]: https://img-blog.csdnimg.cn/92dcc7629105480ba5333fd0bff31d6d.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 11]: https://img-blog.csdnimg.cn/fb565625914c40c4a30ddd760a8b2cad.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[8746899cd8f44a29a3428eae1402e1d3.png]: https://img-blog.csdnimg.cn/8746899cd8f44a29a3428eae1402e1d3.png
[db3dd8357f094de88a6638b586233132.png]: https://img-blog.csdnimg.cn/db3dd8357f094de88a6638b586233132.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 12]: https://img-blog.csdnimg.cn/3470cefe4d924f5fb3b3ca1c75892112.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[3ec62a6fa52e4ff9835cbff95fdd97f5.png]: https://img-blog.csdnimg.cn/3ec62a6fa52e4ff9835cbff95fdd97f5.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 13]: https://img-blog.csdnimg.cn/70678f8f2e2b4cada35a10342351e200.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 14]: https://img-blog.csdnimg.cn/aece351fba944f009e51848f52805530.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[b54d3ac5fc234f35b2b0109efd0eff61.png]: https://img-blog.csdnimg.cn/b54d3ac5fc234f35b2b0109efd0eff61.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 15]: https://img-blog.csdnimg.cn/b23a4ac98f814896b3a1cf96a087aa7c.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[91ec08f8af4c41f6b4811dcac7332543.png]: https://img-blog.csdnimg.cn/91ec08f8af4c41f6b4811dcac7332543.png
[c9f6c4c11978467891d203baa7934eb4.png]: https://img-blog.csdnimg.cn/c9f6c4c11978467891d203baa7934eb4.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_19_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/1b6ddc1627c24bdd894f7dcd6ea473a7.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_19,color_FFFFFF,t_70,g_se,x_16
[9214a4f02c234bd08f4a4845591ffce2.png]: https://img-blog.csdnimg.cn/9214a4f02c234bd08f4a4845591ffce2.png
[ed6e424f0c3945de803c28828487825a.png]: https://img-blog.csdnimg.cn/ed6e424f0c3945de803c28828487825a.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 16]: https://img-blog.csdnimg.cn/b826ee6d8a234da6907ccb32a356bc6f.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 17]: https://img-blog.csdnimg.cn/e3293e4c717b4deb81955656ef8ac9cd.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 18]: https://img-blog.csdnimg.cn/82701a43852c48d5baa685502b4adc4e.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 19]: https://img-blog.csdnimg.cn/4aeceef8c9bf4e02a670bdfe37f24d66.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 20]: https://img-blog.csdnimg.cn/5d204387d7894044bc43fcac528b1a7f.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 21]: https://img-blog.csdnimg.cn/8800ac8cdd7246ceb0f9f7eb238cf761.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 22]: https://img-blog.csdnimg.cn/7dbc4d5baf19469181740def9fca108d.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 23]: https://img-blog.csdnimg.cn/a4c0062a80a5476f9c9f80914564facd.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[07cf18d779684fd381e7919273152547.png]: https://img-blog.csdnimg.cn/07cf18d779684fd381e7919273152547.png