一、前言

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

二、存储架构

每个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

1、接收消息

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

我们现在接着SendMessageProcessor继续看消息是如何持久化的？
在SendMessageProcessor#asyncSendMessage()方法调用MessageStore服务进行持久化消息操作。

接着看MessageStore#asyncPutMessage()方法，其内部将调用putMessage()方法。

由于MessageStore是一个接口，我们看其实现DefaultMessageStore#putMessage()方法。

OK了，入口就是这了：

2、存储消息到CommitLog

1. DefaultMessageStore#putMessage()方法

在上面我们已经介绍过了，Broker端接收完生产消息请求之后会进入到DefaultMessageStore#putMessage()方法。来我们看一下这个putMessage()都做了什么？我们这里将其分为四步进行介绍；

第一步：校验消息存储服务Broker的状态；
包括：DefaultMessageStore消息存储服务状态不能为shutdown，broker的角色必须为mater，消息存储服务是可写的，操作系统页缓存不能是繁忙状态。

第二步：校验该批消息的topic和消息属性信息；
校验信息包括：topic的长度不能大于127B，消息属性的长度不能大于32767B。

第三步：调用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校验；

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

根据事务类型和消息延时发送级别决定是否–设置延时发送的Topic、queueId 并备份真正的topic和queueID。

只有当消息的事务类型为：普通消息、已提交的事务消息时，才会进一步判断消息是否设置了延时级别delayTimeLevel，如果设置了才会走到这个第二步。

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

消息存储的核心步骤就在这里，而且这一步的细节点特别多。我们再分几步进行分析；
（1）首先获取CommitLog中的最后一个MappedFile；

（2）在真正开始写消息之前要对CommitLog加锁，以保证消息串行化写入到CommitLog中；

既然要加锁，那加的是什么锁呢？

我们发现PutMessageLock是一个interface接口，其有两个实现：PutMessageReentrantLock、PutMessageSpinLock。它们一个采用ReentrantLock实现，一个采用CAS机制实现。

那么RocketMQ默认是使用的哪个锁？为什么要使用它？
在CommitLog进行实例化时，会根据MessageStoreConfig配置类中的useReentrantLockWhenPutMessage属性来决定使用哪个锁？
因为useReentrantLockWhenPutMessage的默认值是FALSE，所以使用的是PutMessageSpinLock。
我们可以看到在PutMessageSpinLock中有说明，在低锁竞争时使用它。

（3）设置存储时间戳；对获取到的MappedFile进行校验，确保有一个可用的MappedFile。

如果文件不存在或者已满，会创建一个新的MappedFile文件，并将其添加到mappedFileQueue中。

当需要新建一个MappedFile时，我们来看一下mappedFileQueue.getLastMappedFile(0)做了什么？

它又调用了MappedFileQueue中的一个同名方法，go on, go on，我们继续：

通过获取MappedFiles集合中的最后一个MappedFile的文件名（起始offset）和文件中的消息数量mappedFileSize，拼接出将要创建的MappedFile 和下一个要创建的MappedFile的 文件全路径名；
然后将他们作为一个请求放入到AllocateMappedFileService服务的requestQueue中，由AllocateMappedFileService服务循环进行MMAP操作，分配MappedFile。关于AllocateMappedFileService服务的MMAP操作，我们后面纤细介绍。

（4）向MappedFile中追加消息，追加完消息之后释放锁。

直接向上面获取到的MappedFile追加消息，然后根据返回的状态进一步操作。

如果消息已经写到了MappedFile文件的结尾还没写完，那么就创建一个新的MappedFile文件 重新写入消息；

如果是存放消息失败，直接返回返回错误信息。

追加消息成功之后，统计消息存放消耗的时间。

最后把上面加的锁给释放了。

我们进入MappedFile#appendMessage()方法看一下消息是怎么追加到MappedFile文件的。

追加消息调用的是MappedFile.appendMessage()方法，此方法最终调用到MappedFile.appendMessagesInner()中：

我们来分析下appendMessagesInner()中追加消息的的逻辑：

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

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

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

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

writeBuffer从何而来？

从这里我们可以看到，writeBuffer是在MappedFile初始化的时候从TransientStorePool中获取的。那么，TransientStorePool又是怎么来的？
往上追了一下，没找到，就很尴尬。我想，我想，既然和消息存储有关，会不会在DefaultMessageStore中，呀，真巧，真在呢。
在实例化DefaultMessageStore时，会实例化并初始化transientStorePool。

我们看一下transientStorePool.init()，发现它会实例化poolSize个ByteBuffer放入到双向队列availableBuffers中。

private final Deque<ByteBuffer> availableBuffers;

poolSize的值是多少？

在其构造函数中我们发现poolSize的值取自MessageStoreConfig的transientStorePoolSize变量。

进入MessageStoreConfig类中我们可以看到transientStorePoolSize的默认值为5，所以这个ByteBuffer的池子大小为5。

cb是个啥？从哪来的？

在CommitLog中调用appendMessagesInner()时，传入的 cb 为：this.appendMessageCallback，它的类型为 DefaultAppendMessageCallback，实现了AppendMessageCallback接口。

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

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

>1. 计算消息存储的各个属性，比如：消息长度、消息在ConsumeQueue中的长度等；

>2. 判断消息体大小是否超过4M，超过则返回错误信息；

>3. 判断MappedFile中的可用空间是否可以容纳下要追加的消息，如果不可以，返回错误码END_OF_FILE。
如果返回错误码为END_OF_FILE，接着CommitLog#puMessage()方法中创建一个新的MappedFile，再重新写这条消息。

>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. 返回追加消息成功的结果。

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

即CommitLog数据从ByteBuffer刷到磁盘。详细的我们后面专门介绍。

默认异步刷盘。

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

即主从Broker消息同步。详细的我们后面专门介绍。

默认异步进行主从复制。

第四步：统计消息存储花费的时间；如果存储失败，则增加存储失败的次数；

3、CommitLog刷盘

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

4、后续操作

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

四、总结

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

本文标题：RocketMQ：深入理解消息存储流程

本文链接：https://blog.quwenai.cn/post/9280.html

作者授权：除特别说明外，本文由 toobug 原创编译并授权 TOOBUG信息安全网刊载发布。

本文链接：https://blog.quwenai.cn/post/9280.html

分享到：

正文

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

一、前言

二、存储架构

三、存储流程源码分析

1、接收消息

2、存储消息到CommitLog

1. DefaultMessageStore#putMessage()方法

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

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

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

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

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

3、CommitLog刷盘

4、后续操作

四、总结

相关阅读

【JVM】类加载器（二）：Tomcat 打破双亲委派

MYSQL自增主键丢失问题

【Java杂记】GUI技术：Swing基本使用及JFormDesigner插件安装使用

【Spring】日志：使用 spring 时日志的多种选择

发表评论取消回复

还没有评论，来说两句吧...

目录[+]