RocketMQ Broker对新消息的处理流程
Posted 乐观男孩
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RocketMQ 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宕机,还是会出现部分数据丢失的情况。
以上是关于RocketMQ Broker对新消息的处理流程的主要内容,如果未能解决你的问题,请参考以下文章
RocketMQ源码(11)—Broker asyncPutMessage处理消息以及存储的高性能设计一万字