RocketMQ Broker对新消息的处理流程

Posted 2022-03-21 乐观男孩

目录

• 处理流程图
• Broker启动入口
• 组件初始化过程
• 消息处理过程
• 消息刷新到磁盘
• 总结

处理流程图

NettyServer：RocketMQ基于Netty服务器，NettyServer用于接收Client的请求（这里是新消息请求）。
SendMessageProcessor：对请求进行预处理（解析请求命令、解析请求内容等）
DefaultMessageStore：默认消息存储器，实现关于消息的存取操作。
CommitLog：具体处理消息的存储逻辑，其内部维护了一些里的MapperFile。
MapperFile：消息最终会存储在磁盘的一个个文件上，MapperFile就是这些磁盘文件在程序中的一份映射，对MapperFile的读写就相当于对磁盘的读写（当然，内部会涉及一些数据落盘的刷新操作）。

Broker启动入口

从broker启动脚本（bin目录下的mqbroker文件）中，可以知道Broker的启动类是BrokerStartup：

BrokerStartup创建BrokerController并进行启动，BrokerController又会对一系列的组件进行初始化及启动。

组件初始化过程

在了解新消息处理流程之前，先了解其中涉及组件的组装过程，以便对其中的组件初始化有一个大概的认识。
BrokerController的initialize()方法中，会实例化DefaultMessageStore和NettyRemotingServer：


BrokerController的registerProcessor()方法，为NettyRemotingServer针对不同的请求类型绑定不同的请求处理器（如针对SEND_MESSAGE的请求类型，就绑定了SendMessageProcessor请求处理器）

BrokerController对DefaultMessageStore进行实例化，DefaultMessageStore的构造函数中，会实例化CommitLog实例

在CommitLog的构造函数中，会实例化MappedFileQueue（维护一系列MapperFile）和FlushRealTimeService（负责将内存中的数据刷新到磁盘文件）

MappedFileQueue具备加载磁盘中的MapperFile和创建新的MapperFile能力

消息处理过程

从上面介绍可以知道，新消息的请求首先被NettyRemotingServer进行处理。在NettyRemotingServer进行启动时，会将NettyServerHandler加入的处理器链中，由NettyServerHandler实例处理请求

NettyServerHandler实际调用processMessageReceived()方法进行处理

然后继续跟踪链路调用到processRequestCommand()方法，该方法会根据具体的请求类型获取具体的请求处理器（请求处理器已在BrokerController的registerProcessor()方法中进行注册）

然后请求会交由SendMessageProcessor进行处理。SendMessageProcessor会对请求的内容进行解析，并封装成消息的形式（MessageExtBrokerInner），将消息提交给DefaultMessageStore进行处理

DefaultMessageStore将消息提交给CommitLog进行处理

CommitLog对消息的处理，是处理消息的核心流程，处理内容主要包括几方面：对新消息按照存储的格式进行编码、获取最后一个MapperFile、将消息追加到MapperFile上
1、对新消息按照存储的格式进行编码

编码，其实就是按照Broker存储的消息格式进行消息组装，组装后的内容存在encodeBuffer中（此时encodeBuffer的某些项其实是初始值，需要在后续的AppendMessageCallback.doAppend()过程中填充）

Broker存储的消息格式

2、获取最后一个MapperFile，其实就是获取一个可以写入消息的MapperFile，如果最新的MapperFile已经写满（1G），则会创建一个新的MapperFile

3、将消息追加到MapperFile上，是通过调用MapperFile的appendMessage()方法，具体有用逻辑是在MapperFile.appendMessagesInner()方法上

AppendMessageCallback.doAppend()方法，实际是对EncodedBuff中的某些项进行填充，填充完后，EncodedBuff里的内容就是按照Broker存储格式的完整消息，然后会将EncodedBuff的内容转移到MapperFile的MappedByteBuffer中

这里有几点需要着重说明：
（1）、MapperFile，实际是磁盘文件在应用的一份映射，文件内容是虚拟内存的方式映射到MapperFile的MappedByteBuffer（以MappedByteBuffer进行讲解）中。
（2）、新消息的写入，实际是写到MapperFile的MappedByteBuffer中。
（3）、RocketMQ针对MapperFile的MappedByteBuffer，通过wrotePosition单独维护写入的位置，这样可以通过对MappedByteBuffer的slice操作，获取子buffer，可以做很多灵活的操作。
（4）、MapperFile通过MappedByteBuffer，使用顺序写的方式，提高了读写速度和降低内存占用。
针对以上几点，用几张图辅助理解：
a.磁盘文件内容映射在MapperFile的MappedByteBuffer中：

b.新消息写入MappedByteBuffer变化图示

消息刷新到磁盘

在CommitLog.asyncPutMessage()方法内调用完MappedFile.appendMessage()方法后，新消息实际只追加到MapperFile的MappedByteBuffer中，但未实际保存到磁盘。接着CommitLog.asyncPutMessage()方法会执行消息刷新磁盘（submitFlushRequest()）和消息复制到从Broker（submitReplicaRequest()）的操作

从submitFlushRequest()方法可以知道，有两种处理方式：
1、如果配置成同步刷盘机制，分两种情况处理：
a、客户端配置成等待消息保存成功（messageExt.isWaitStoreMsgOK()）才返回，则每一条新消息追加到MapperFile的MappedByteBuffer后，都会执行写磁盘操作，写成功才返回；
b、客户端没有配置成等待消息保存成功（messageExt.isWaitStoreMsgOK()）才返回，此时唤醒flushCommitLogService（GroupCommitService类型）。
2、异步刷盘机制：直接唤醒flushCommitLogService（FlushRealTimeService类型）或commitLogService（类型CommitRealTimeService，这个后面就不分析了，属于旧的api）。异步刷盘是综合考虑了性能和安全性，一般都是配置成异步刷新。这种方式实际上只是唤醒了FlushRealTimeService就直接返回PutMessageStatus.PUT_OK了，其实也没有执行具体刷盘操作。

接下来看一下消息同步到从Broker（submitReplicaRequest()）的操作，也分两种方式处理：
1、如果Broker配置成同步复制（BrokerRole.SYNC_MASTER），分两种情况处理：
a、客户端配置成等待复制成功（messageExt.isWaitStoreMsgOK()）才返回，则每一条新消息追加到MapperFile的MappedByteBuffer后，都会执行复制操作，复制成功才返回；
b、直接返回PutMessageStatus.PUT_OK。
2、如果Broker配置成异步复制（BrokerRole.ASYNC_MASTER），则直接返回PutMessageStatus.PUT_OK。默认配置就是异步复制。
总结：综上，从默认配置（异步刷盘、异步复制）上看，执行完submitFlushRequest()方法后，实际也没有把消息刷新到磁盘，只是唤醒了FlushRealTimeService服务。执行完submitReplicaRequest()方法后，也没有实际将消息复制到从Broker，只是直接返回了PutMessageStatus.PUT_OK，没有唤醒其他服务。
接下来，我们看看FlushRealTimeService服务具体的功能是什么。

FlushRealTimeService是一个线程类服务，所以直接看run()方法即可。从run方法可以知道，其调用了MappedFileQueue.flush()方法

MappedFileQueue.flush()根据flushedWhere标识找到对应的MapperFile，然后调用MapperFile.flush()方法，该方法直接将MapperFile的MappedByteBuffer刷新到磁盘，完成持久化操作。

补充两张张关于flushedWhere的说明图
FlushRealTimeService.run()执行前：

FlushRealTimeService.run()执行后：

总结

1、针对Broker服务端对请求的处理，服务端会对不同请求绑定不同的请求处理器（如新消息处理器就是SendMessageProcessor）。请求处理器解析请求内容，封装成不同的数据，交由后续流程进行业务处理。
2、所有消息不区分Topic，都写在同一个磁盘文件中。一个磁盘大小默认为1GB，达到指定大小后，将生成新文件进行存储。每个磁盘文件，在应用程序中，都对应一个MapperFile。在MapperFile内，有一个MappedByteBuffer字节数组存储文件的具体内容。在写入时，是顺序对MappedByteBuffer写入，从而实现顺序写此盘的机制，提升了写入速度；另外，使用MappedByteBuffer，减少了数据的之间的拷贝，也大大提升了写入效率。
3、Broker对MappedByteBuffer的读写，不会直接操作MappedByteBuffer，而是通过维护wrotePosition、flushWhere等标识，通过slice操作生成子Buffer间接完成对MappedByteBuffer的读写，这样的好处是灵活性非常高。
4、新消息写入，先写入MapperFile的MappedByteBuffer，然后通过刷盘机制（同步、异步），将MappedByteBuffer的消息刷新到磁盘中。
5、从消息刷盘机制可以看出，如果是异步刷盘方式，在新消息写入到Mapper的MappedByteBuffer后，Broker就会返回成功给Producer，消息会在后续异步刷新到磁盘。如果在返回成功给Producer后，消息刷盘前Broker宕机，还是会出现部分数据丢失的情况。