从源码分析RocketMQ系列-RocketMQ消息持久化源码详解

r囧r小猫 2023-07-18 09:21 13阅读 0赞

> **导语**  
>   在上篇分析中，提到了一个概念处理器，并且在进入到最终NettyIO的时候看到了一个Pair的对象，这个对象存储了两个对象，一个是执行器，一个是处理器，在进入Runable对象的时候看到封装到Pair对象中的一个处理器NettyRequestProcessor，这个处理器被称为是请求处理器。下面就来详细的进入请求处理器的世界

### 文章目录 ###

*   *  NettyRequestProcessor
     *   *  SendMessageProcessor
         *  封装操作
         *   *  消息持久化操作
             *   *  1.1 消息存储整体架构
                 *  1.2 页缓存与内存映射
                 *  1.3 消息刷盘
             *  源码分析
             *   *  持久化逻辑
                 *  刷盘逻辑
     *  总结

## NettyRequestProcessor ##

这个是一个是一个接口类，具体代码如下

/** * Common remoting command processor */
    public interface NettyRequestProcessor { 
        RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
            throws Exception;
    
        boolean rejectRequest();
    
    }

通用远程处理命令处理器，其中有两个方法一个是拒绝请求，一个是处理请求，在处理请求的方法中有两个参数

*  ChannelHandlerContext Channel处理器的上下文信息
 *  RemotingCommand ：封装的请求参数

如下图所示，看到在RocketMQ中很多地方都继承了这个接口并且实现了这个接口中的方法。从Broker模块到NameServer模块以及Remoting模块都涉及到了，当然这个处理类出现在不同的模块中它所包含的功能也是不一样的下面就来按照模块以及功能分析一下这些处理器。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70]

### SendMessageProcessor ###

发送消息处理器，根据上面的接口逻辑，找到processRequest()方法，方法中调用了一个asyncProcessRequest(ctx, request).get()方法，是一个链式调用首先调用了asyncProcessRequest(ctx, request)方法然后从该方法的返回值中获取到了Response对象。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 1]  
  下面就来进入到asyncProcessRequest(ctx, request)方法中来看看  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 2]  
  进入方法中先进行了RequestCode的判断CONSUMER\_SEND\_MSG\_BACK，表示消费者发送消息。默认执行了后续的逻辑。

*  1、先处理了Request携带的请求头信息
 *  2、从获取到的请求头信息中中封装一个消息传输的上下文操作，这个上下文是由传入的Channel上下文和请求头一起组成的。
 *  3、判断消息是否是批处理的，这里来看看非批处理的操作。return this.asyncSendMessage(ctx, request, mqtraceContext, requestHeader);

通过上面三个步骤就进入了同步发送消息的方法中。经过一系列的封装之后操作方法最终落到了一个handlePutMessageResultFuture()方法上。它的返回值为一个CompletableFuture 对象。这个就表示，在执行完成之后我们的操作想获取这样的一个对象，但是实际上可能由于其他问题，并不能得到这个对象。在很大程度上这个操作是可以完成的，这样的话，程序执行到这里就不需要等待结果了，而是从这里获取到了一个结果，这个结果在后面的逻辑中会进行组装。这种方式被称为是Future模式。

return handlePutMessageResultFuture(putMessageResult, response, request, msgInner, responseHeader, mqtraceContext, ctx, queueIdInt);

进入到上面方法中之后执行了下面下面这个操作，

return putMessageResult.thenApply((r) ->
                handlePutMessageResult(r, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt)

上面提到了Future模式在多线程中Future模式也是比较常用的模式，这个就类似于去干洗店洗衣服，但是干洗店想要告诉你的是你当时并不能拿到衣服，就给了你一个凭条，后续的话你可以在你方便的时候，拿着这个凭条去干洗店取自己的衣服。而这里的CompletableFuture这个对象就类似于这样一个凭条。

在程序调用handlePutMessageResult方法的时候默认其实已经拿到了个对应的Future对象。所以说程序到这个方法的时候就已经结束了。后续或者是前面的操作都是对这个方法的调用参数以及返回参数的封装操作，方便后面的其他操作。

### 封装操作 ###

在上面的分析中，我们知道了在上面提到的两个方法前后都是对结果的封装操作，在RocketMQ中有两个比较重要的概念就是消息的事务管理以及消息的持久化操作。在调用上面方法的时候回发现在操作CompletableFuture 对象的时候有如下的一段代码。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 3]  
 &emps;在代码中有两个比较关键的操作就是事务操作和消息持久化的操作，这里来看看这两个操作分别是在什么条件下触发的。

**事务操作与持久化判断条件**  
  进入操作之前先有一个判断transFlag != null && Boolean.parseBoolean(transFlag) 这个判断决定了是否进行持久化或者选择事务操作。如果条件成立，这进入到事务操作中this.brokerController.getBrokerConfig().isRejectTransactionMessage()并且判断是否配置了拒绝事务的操作。这里先来看一下消息的持久化操作。

putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);

#### 消息持久化操作 ####

消息存储是RocketMQ中最复杂、最重要的部分。介绍RocketMQ的三个方面：

*  消息存储体系结构
 *  页面缓存和内存映射
 *  RocketMQ的两种不同的磁盘刷新方法。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 4]

消息存储是RocketMQ中最为复杂和最为重要的一部分，本节将分别从RocketMQ的消息存储整体架构、PageCache与Mmap内存映射以及RocketMQ中两种不同的刷盘方式三方面来分别展开叙述。

##### 1.1 消息存储整体架构 #####

消息存储架构图中主要有下面三个跟消息存储相关的文件构成。

*  (1) CommitLog：消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G ，文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件；
 *  (2) ConsumeQueue：消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。consumequeue文件可以看成是基于topic的commitlog索引文件，故consumequeue文件夹的组织方式如下：topic/queue/file三层组织结构，具体存储路径为：$HOME/store/consumequeue/\{topic\}/\{queueId\}/\{fileName\}。同样consumequeue文件采取定长设计，每一个条目共20个字节，分别为8字节的commitlog物理偏移量、4字节的消息长度、8字节tag hashcode，单个文件由30W个条目组成，可以像数组一样随机访问每一个条目，每个ConsumeQueue文件大小约5.72M；
 *  (3) IndexFile：IndexFile（索引文件）提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是：$HOME \\store\\index$\{fileName\}，文件名fileName是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现HashMap结构，故rocketmq的索引文件其底层实现为hash索引。

在上面的RocketMQ的消息存储整体架构图中可以看出，RocketMQ采用的是混合型的存储结构，即为Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。RocketMQ的混合型存储结构(多个Topic的消息实体内容都存储于一个CommitLog中)针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构，Producer发送消息至Broker端，然后Broker端使用同步或者异步的方式对消息刷盘持久化，保存至CommitLog中。只要消息被刷盘持久化至磁盘文件CommitLog中，那么Producer发送的消息就不会丢失。正因为如此，Consumer也就肯定有机会去消费这条消息。当无法拉取到消息后，可以等下一次消息拉取，同时服务端也支持长轮询模式，如果一个消息拉取请求未拉取到消息，Broker允许等待30s的时间，只要这段时间内有新消息到达，将直接返回给消费端。这里，RocketMQ的具体做法是，使用Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue（逻辑消费队列）和IndexFile（索引文件）数据。

##### 1.2 页缓存与内存映射 #####

页缓存（PageCache)是OS对文件的缓存，用于加速对文件的读写。一般来说，程序对文件进行顺序读写的速度几乎接近于内存的读写速度，主要原因就是由于OS使用PageCache机制对读写访问操作进行了性能优化，将一部分的内存用作PageCache。对于数据的写入，OS会先写入至Cache内，随后通过异步的方式由pdflush内核线程将Cache内的数据刷盘至物理磁盘上。对于数据的读取，如果一次读取文件时出现未命中PageCache的情况，OS从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取。

在RocketMQ中，ConsumeQueue逻辑消费队列存储的数据较少，并且是顺序读取，在page cache机制的预读取作用下，Consume Queue文件的读性能几乎接近读内存，即使在有消息堆积情况下也不会影响性能。而对于CommitLog消息存储的日志数据文件来说，读取消息内容时候会产生较多的随机访问读取，严重影响性能。如果选择合适的系统IO调度算法，比如设置调度算法为“Deadline”（此时块存储采用SSD的话），随机读的性能也会有所提升。

另外，RocketMQ主要通过MappedByteBuffer对文件进行读写操作。其中，利用了NIO中的FileChannel模型将磁盘上的物理文件直接映射到用户态的内存地址中（这种Mmap的方式减少了传统IO将磁盘文件数据在操作系统内核地址空间的缓冲区和用户应用程序地址空间的缓冲区之间来回进行拷贝的性能开销），将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率（正因为需要使用内存映射机制，故RocketMQ的文件存储都使用定长结构来存储，方便一次将整个文件映射至内存）。

##### 1.3 消息刷盘 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 5]

*  (1) 同步刷盘：如上图所示，只有在消息真正持久化至磁盘后RocketMQ的Broker端才会真正返回给Producer端一个成功的ACK响应。同步刷盘对MQ消息可靠性来说是一种不错的保障，但是性能上会有较大影响，一般适用于金融业务应用该模式较多。
 *  (2) 异步刷盘：能够充分利用OS的PageCache的优势，只要消息写入PageCache即可将成功的ACK返回给Producer端。消息刷盘采用后台异步线程提交的方式进行，降低了读写延迟，提高了MQ的性能和吞吐量。

#### 源码分析 ####

##### 持久化逻辑 #####

上面的内容结合官方文档对整个的概念做了分析，下面就来接着上面概念从源码的角度上来进行分析。首先从上面的源码中找到了了如下的一些方法。这些方法共同的组成了整个的消息持久化机制。

public MessageStore getMessageStore() { 
        return messageStore;
     }
     
     default CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) { 
        return CompletableFuture.completedFuture(putMessage(msg));
    }

下面这个方法看上去应该是消息持久化的核心方法，在进入commitlog.putMessage()方法之前首先做了两个判断操作一个是消息被消费了另一个是消息被忽略了。最后这两个操作都没有进行就会进入到putMessage的逻辑中。而在这个逻辑中完成的操作就是所描述的操作，而这个方法是存在于DefaultMessageStore这类中，也就是说所有的关于上面的操作操作包括CommitLog 、ConsumerQueue等等的信息都是直接由这个类来进行的操作。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 6]  
  会发现在该方法中有一个如下的操作，这个操作也许是隐藏在所有操作中最为重要的方法。这个的代码到这里才执行到将上面图中的Producer发送的消息放入到了CommitLog中。CommitLog的putMessage()方法中所展示的逻辑就是上面文字描述的逻辑。

PutMessageResult result = this.commitLog.putMessage(msg);

##### 刷盘逻辑 #####

在放入的CommitLog的putMessage()方法的时候，其逻辑与上面的逻辑相同。但是在执行到之后的时候分别调用了两个方法。分别表示刷盘和高可用。

handleDiskFlush(result, putMessageResult, msg);
    handleHA(result, putMessageResult, msg);

进入到handleDiskFlush(result, putMessageResult, msg);方法中会看到这里与上面的逻辑一样提供了两种刷盘逻辑，一种是同步刷盘逻辑，一种是异步刷盘逻辑。看到同步刷盘逻辑中的逻辑较异步刷盘逻辑中的较多。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 7]  
  从上面的代码逻辑课可以看到，实际处理同步异步的并不是这个方法，而是有flushCommitLogService这样的一个服务，而这个服务是一个具体调用要看它实际最后被那个子类实现，在上面的方法中可以看到在同步刷盘策略中的对象被强制类型转换成了GroupCommitService，而最终使用的方法是下面的两个方法，但是会看到在后面的代码中也调用了wakeup()方法，可以推测，这个方法是父类方法。不需要关注，关注的分别是这几个服务中run()方法，

service.putRequest(request); 
    
    service.wakeup();

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 8]  
  上图展示了在GroupCommitService中的putRequest()方法调用。

也就是说FlushCommitLogService类其实，是实现了三个不一样的类，分别来适配三个操作

*  GroupCommitService 同步调用的服务类。
 *  CommitRealTimeService 异步调用 ，在异步调用中commitlogService的支持
 *  FlushRealTimeService 异步调用，在异步调用中 实际支持flushCommitLogService

## 总结 ##

到这里其实已经进入到了线程级别的分析，分别需要支持两种刷盘策略就是执行了两种不一样的线程策略，不同的线程执行策略，带来的效果是不一样的。同时在分析方法的过程中也看到了很多的配置相关的选择也进入到了判断中。对于这些线程级别的IO分析。在后面的分析中会提到，到时候就需要有跟多的关于多线程知识的掌握。请大家关注博主的多线程相关的分析。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70]: /images/20230528/68e3c489fb4a463abe0099fe5351fa6f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 1]: /images/20230528/734f7a6eb6534813aba0d6d281fd3a32.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 2]: /images/20230528/5a128a17cb774faa9abcf836dd207c1d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 3]: /images/20230528/c564bc9133624a5089e183b213cd87c3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 4]: /images/20230528/9a6e298975f446bfb54a149771cdfb4d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 5]: /images/20230528/4daf2d5c52e842cabf0b71ae28ebd0b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 6]: /images/20230528/75b303fff50b43f69a28c91e39531bf2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 7]: /images/20230528/dca1e2d65bf144fb826e5346bdbd1175.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 8]: /images/20230528/a8e4eeeb690d4d7693a7d4c4a4876867.png