从本文开始，我们来分析rocketMq消息接收、分发以及投递流程。
RocketMq消息处理整个流程如下：
- 消息接收：消息接收是指接收producer的消息，处理类是SendMessageProcessor，将消息写入到commigLog文件后，接收流程处理完毕；
- 消息分发：broker处理消息分发的类是ReputMessageService，它会启动一个线程，不断地将commitLong分到到对应的consumerQueue，这一步操作会写两个文件：consumerQueue与indexFile，写入后，消息分发流程处理 完毕；
- 消息投递：消息投递是指将消息发往consumer的流程，consumer会发起获取消息的请求，broker收到请求后，调用PullMessageProcessor类处理，从consumerQueue文件获取消息，返回给consumer后，投递流程处理完毕。
以上就是rocketMq处理消息的流程了，接下来我们就从源码来看相关流程的实现。
1. remotingServer的启动流程
在正式分析接收与投递流程前，我们来了解下remotingServer的启动。
remotingServer是一个netty服务，他开启了一个端口用来处理producer与consumer的网络请求。
remotingServer是在BrokerController#start中启动的，代码如下：
public void start() throws Exception {
// 启动各组件
...
if (this.remotingServer != null) {
this.remotingServer.start();
}
...
}
继续查看remotingServer的启动流程，进入NettyRemotingServer#start方法：
public void start() {
...
ServerBootstrap childHandler =
this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
...
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast(defaultEventExecutorGroup,
HANDSHAKE_HANDLER_NAME, handshakeHandler)
.addLast(defaultEventExecutorGroup,
encoder,
new NettyDecoder(),
new IdleStateHandler(0, 0,
nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
connectionManageHandler,
// 处理业务请求的handler
serverHandler
);
}
});
...
}
这就是一个标准的netty服务启动流程了，套路与nameServer的启动是一样的。关于netty的相关内容，这里我们仅关注pipeline上的channelHandler，在netty中，处理读写请求的操作为一个个ChannelHandler，remotingServer中处理读写请求的ChanelHandler为NettyServerHandler，代码如下：
@ChannelHandler.Sharable
class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
processMessageReceived(ctx, msg);
}
}
这块的操作与nameServer对外提供的服务极相似（就是同一个类），最终调用的是NettyRemotingAbstract#processRequestCommand方法：
public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
// 根据 code 从 processorTable 获取 Pair
final Pair<NettyRequestProcessor, ExecutorService> matched
= this.processorTable.get(cmd.getCode());
// 找不到默认值
final Pair<NettyRequestProcessor, ExecutorService> pair =
null == matched ? this.defaultRequestProcessor : matched;
...
// 从 pair 中拿到 Processor 进行处理
NettyRequestProcessor processor = pair.getObject1();
// 处理请求
RemotingCommand response = processor.processRequest(ctx, cmd);
....
}
如果进入源码去看，会发现这个方法非常长，这里省略了异步处理、异常处理及返回值构造等，仅列出了关键步骤：
- 根据code从processorTable拿到对应的Pair
- 从Pair里获取Processor
最终处理请求的就是Processor了。
2. Processor的注册
从上面的分析中可知， Processor是处理消息的关键，它是从processorTable中获取的，这个processorTable是啥呢？
processorTable是NettyRemotingAbstract成员变量，里面的内容是BrokerController在初始化时（执行BrokerController#initialize方法）注册的。之前在分析BrokerController的初始化流程时，就提到过Processor的提供操作，这里再回顾下：
BrokerController的初始化方法initialize会调用 BrokerController#registerProcessor，Processor的注册操作就在这个方法里：
public class BrokerController {
private final PullMessageProcessor pullMessageProcessor;
/**
* 构造方法
*/
public BrokerController(...) {
// 处理 consumer 拉消息请求的
this.pullMessageProcessor = new PullMessageProcessor(this);
}
/**
* 注册操作
*/
public void registerProcessor() {
// SendMessageProcessor
SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
sendProcessor.registerSendMessageHook(sendMessageHookList);
sendProcessor.registerConsumeMessageHook(consumeMessageHookList);
// 处理 Processor
this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE,
sendProcessor, this.sendMessageExecutor);
this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2,
sendProcessor, this.sendMessageExecutor);
this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE,
sendProcessor, this.sendMessageExecutor);
// PullMessageProcessor
this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE,
this.pullMessageProcessor, this.pullMessageExecutor);
// 省略其他许许多多的Processor注册
...
}
...
需要指明的是，sendProcessor用来处理producer请求过来的消息，pullMessageProcessor用来处理consumer拉取消息的请求。
3. 接收producer消息
了解完remotingServer的启动与Processor的注册内容后，接下来我们就可以分析接收producer消息的流程了。
producer发送消息到broker时，发送的请求code为SEND_MESSAGE（RocketMQ源码5-producer 同步发送和单向发送 第1.4小节），根据上面的分析，当消息过来时，会使用NettyServerHandler这个ChannelHandler来处理，之后会调用到NettyRemotingAbstract#processRequestCommand方法。
在NettyRemotingAbstract#processRequestCommand方法中，会根据消息的code获取对应的Processor来处理，从Processor的注册流程来看，处理该SEND_MESSAGE的Processor为SendMessageProcessor，我们进入SendMessageProcessor#processRequest看看它的流程：
public RemotingCommand processRequest(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
RemotingCommand response = null;
try {
// broker处理接收消息
response = asyncProcessRequest(ctx, request).get();
} catch (InterruptedException | ExecutionException e) {
log.error("process SendMessage error, request : " + request.toString(), e);
}
return response;
}
没干啥事，一路跟下去，直接看普通消息的流程，进入SendMessageProcessor#asyncSendMessage方法：
private CompletableFuture<RemotingCommand> asyncSendMessage(ChannelHandlerContext ctx,
RemotingCommand request, SendMessageContext mqtraceContext,
SendMessageRequestHeader requestHeader) {
final RemotingCommand response = preSend(ctx, request, requestHeader);
final SendMessageResponseHeader responseHeader
= (SendMessageResponseHeader)response.readCustomHeader();
if (response.getCode() != -1) {
return CompletableFuture.completedFuture(response);
}
final byte[] body = request.getBody();
int queueIdInt = requestHeader.getQueueId();
TopicConfig topicConfig = this.brokerController.getTopicConfigManager()
.selectTopicConfig(requestHeader.getTopic());
// 如果没指定队列，就随机指定一个队列
if (queueIdInt < 0) {
queueIdInt = randomQueueId(topicConfig.getWriteQueueNums());
}
// 将消息包装为 MessageExtBrokerInner
MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
msgInner.setTopic(requestHeader.getTopic());
msgInner.setQueueId(queueIdInt);
// 省略处理 msgInner 的流程
...
CompletableFuture<PutMessageResult> putMessageResult = null;
Map<String, String> origProps = MessageDecoder
.string2messageProperties(requestHeader.getProperties());
String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
// 发送事务消息
if (transFlag != null && Boolean.parseBoolean(transFlag)) {
...
// 发送事务消息
putMessageResult = this.brokerController.getTransactionalMessageService()
.asyncPrepareMessage(msgInner);
} else {
// 发送普通消息
putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);
}
return handlePutMessageResultFuture(putMessageResult, response, request, msgInner,
responseHeader, mqtraceContext, ctx, queueIdInt);
}
这个方法是在准备消息的发送数据，所做的工作如下：
- 如果没指定队列，就随机指定一个队列，一般情况下不会给消息指定队列的，但如果要发送顺序消息，就需要指定队列了，这点后面再分析。
- 构造MessageExtBrokerInner对象，就是将producer上送的消息包装下，加上一些额外的信息，如消息标识msgId、发送时间、topic、queue等。
- 发送消息，这里只是分为两类：事务消息与普通消息，这里我们主要关注普通消息，事务消息后面再分析。
进入普通消息的发送方法DefaultMessageStore#asyncPutMessage：
public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {
...
// 保存到 commitLog
CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);
...
}
3.1 commitLog写入原理
一个broker逻辑上对应着一个commitLog，你可以把它看作一个大文件，然后这个broker收到的所有消息都写到这个里面，但是物理上ROCKET_HOME/commitlog/00000000000000000000这个路径存储的，它是由若干个文件组成的，每个文件默认大小是1G，然后每个文件都对应这个一个MappedFile，00000000000000000000 就是第一个MappedFile对应的物理文件，每个文件的文件名就是在commitLog里面的一个其实offset，第二个文件名就是00000000001073741824，也就是上一个MappedFile文件起始offset加上每个文件的大小，这个MappedFile就是RocketMQ的黑科技，使用了内存映射技术来提高文件的访问速度与写入速度，然后都是采用追加写的方式提高写入速度。
我们直接看官方的描述(链接：github.com/apache/rock…)：
rocketMq 消息存储架构图
消息存储架构图中主要有下面三个跟消息存储相关的文件构成。
(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（索引文件）数据。
3.2 CommitLog#asyncPutMessage
继续进入CommitLog#asyncPutMessage方法，这个方法有点长，我们分部分：
public CompletableFuture<PutMessageResult> asyncPutMessage(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());
// 事务
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 = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
// Backup real topic, queueId
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);
}
}
...
这一部分其实就是从msg中获取一些信息，判断处理一下这个延时消息。
long elapsedTimeInLock = 0;
MappedFile unlockMappedFile = null;
// 获取最后一个 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 是否是 null 或者是否是满了
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 CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null));
}
// todo 往mappedFile追加消息
result = mappedFile.appendMessage(msg, this.appendMessageCallback);
...
这一部分就比较重要了，首先是从mappedFileQueue中获取最后一个MappedFile，这个就是拿集合最后一个元素，因为都是有序的，最后一个元素就是最后一个MappedFile，接着就是获取锁了，这个锁也是比较有讲究的，可以设置使用ReentrantLock 也可以设置使用cas，默认是使用cas，接着就是设置beginTimeInLock这个变量了，这个变量我们在判断os page cache繁忙的时候说过，就是获取到锁的一个时间戳，在释放锁之前会重置成0，接着就是判断mappedFile是不是null或者是不是满了，如果是的话就要新建一个了。
接着就是最最最重要的了 往mappedFile中追加消息，
mappedFile.appendMessage
/**
* 将消息追加到MappedFile文件中
*/
public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb) {
assert messageExt != null;
assert cb != null;
// 获取MappedFile当前文件写指针
int currentPos = this.wrotePosition.get();
// 如果currentPos小于文件大小
if (currentPos < this.fileSize) {
ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
byteBuffer.position(currentPos);
AppendMessageResult result;
// 单个消息
if (messageExt instanceof MessageExtBrokerInner) {
// todo
result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt);
// 批量消息
} else if (messageExt instanceof MessageExtBatch) {
result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt);
} else {
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}
this.wrotePosition.addAndGet(result.getWroteBytes());
this.storeTimestamp = result.getStoreTimestamp();
return result;
}
// 如果currentPos大于或等于文件大小，表明文件已写满，抛出异常
log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}
这里首先获取了一下这个mappedFile写到哪个位置了，它这个位置是从0开始的，然后判断一下当前位置与文件大小做对比，要是大于的就超了文件大小了，接着是获取writerbuffer因为这里是没有开启transientStorePool的，所以它是个空的，就会使用mmapedByteBuffer，接着就是调用回调的doAppend追加消息了，我们看下它的参数， 第一个是开始offset，这个offset是commitlog的一个offset，举个例子，第一个MappedFile的开始offset是0，然后一个MappedFile 的大小是1g，然后第二个MappedFile就得从1073741824（1g）开始了，第二个参数是bytebuffer，这个不用多说，第三个是这个MappedFile还空多少字节没用，第四个就是消息了。
我们来看下这个doAppend方法，这个也有点长，我们需要分开看下：
/**
* // 只是将消息追加到内存中
* @param fileFromOffset 文件的第一个偏移量（就是MappedFile是从哪个地方开始的）
*/
public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
final MessageExtBrokerInner msgInner) {
// STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>
// PHY OFFSET
long wroteOffset = fileFromOffset + byteBuffer.position();
int sysflag = msgInner.getSysFlag();
int bornHostLength = (sysflag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
int storeHostLength = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength);
ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
this.resetByteBuffer(storeHostHolder, storeHostLength);
// 创建全局唯一消息id
String msgId;
if ((sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
} else {
msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
}
// Record ConsumeQueue information
keyBuilder.setLength(0);
keyBuilder.append(msgInner.getTopic());
keyBuilder.append('-');
keyBuilder.append(msgInner.getQueueId());
String key = keyBuilder.toString();
// 获取该消息在消息队列的物理偏移量
Long queueOffset = CommitLog.this.topicQueueTable.get(key);
if (null == queueOffset) {
queueOffset = 0L;
CommitLog.this.topicQueueTable.put(key, queueOffset);
}
// Transaction messages that require special handling
final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
switch (tranType) {
// Prepared and Rollback message is not consumed, will not enter the
// consumer queue
case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
queueOffset = 0L;
break;
case MessageSysFlag.TRANSACTION_NOT_TYPE:
case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
default:
break;
}
这一部分主要就是 计算了一下这个消息写在commitlog中的一个offset，接着就是生成一个msgId，然后根据topic 与queueId从缓存中获取了一下这个queueId对应的一个queue的offset，这个其实就是添加一个消息加1，然后就是事务的东西了，如果有事务，然后还在准备阶段或者回滚阶段，就将queue offset 设置成0，再往下其实就是处理消息，然后写到buffer中了。
/**
* Serialize message
*/
final byte[] propertiesData =
msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
final int propertiesLength = propertiesData == null ? 0 : propertiesData.length;
if (propertiesLength > Short.MAX_VALUE) {
log.warn("putMessage message properties length too long. length={}", propertiesData.length);
return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
}
final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
final int topicLength = topicData.length;
final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
// todo 计算消息总长度
final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);
// Exceeds the maximum message
if (msgLen > this.maxMessageSize) { // 最大消息长度 65536
CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLength
+ ", maxMessageSize: " + this.maxMessageSize);
return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
}
...
这里首先是获取了一下消息里面的properties，将它转成字节数组，计算了一下长度，接着就是将topic转成字节数据，计算了一下长度，获取了一下body的长度，就是你往Message塞得内容长度，重点来，计算 消息的总长度，然后判断一下长度是否超长。其中calMsgLength如下：
// 计算消息长度
protected static int calMsgLength(int sysFlag, int bodyLength, int topicLength, int propertiesLength) {
int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
int storehostAddressLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 8 : 20;
final int msgLen = 4 //TOTALSIZE 消息条目总长度，4字节
+ 4 //MAGICCODE 魔数，4字节。固定值0xdaa320a7
+ 4 //BODYCRC 消息体的crc校验码，4字节
+ 4 //QUEUEID 消息消费队列ID，4字节
+ 4 //FLAG 消息标记，RocketMQ对其不做处理，供应用程序使用， 默认4字节
+ 8 //QUEUEOFFSET 消息在ConsumeQuene文件中的物理偏移量，8字节。
+ 8 //PHYSICALOFFSET 消息在CommitLog文件中的物理偏移量，8字节
+ 4 //SYSFLAG 消息系统标记，例如是否压缩、是否是事务消息 等，4字节
+ 8 //BORNTIMESTAMP 消息生产者调用消息发送API的时间戳，8字 节
+ bornhostLength //BORNHOST 消息发送者IP、端口号，8字节
+ 8 //STORETIMESTAMP 消息存储时间戳，8字节
+ storehostAddressLength //STOREHOSTADDRESS Broker服务器IP+端口号，8字节
+ 4 //RECONSUMETIMES 消息重试次数，4字节
+ 8 //Prepared Transaction Offset 事务消息的物理偏移量，8 字节。
+ 4 // 消息体长度，4字节
+ (bodyLength > 0 ? bodyLength : 0) // BODY 消息体内容，长度为bodyLenth中存储的值
+ 1 // 主题存储长度，1字节，表示主题名称不能超过255个字符。
+ topicLength //TOPIC 主题，长度为TopicLength中存储的值
+ 2 // 消息属性长度，2字节，表示消息属性长 度不能超过65536个字符。
+ (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength 消息属性，长度为PropertiesLength中存储的 值
+ 0;
return msgLen;
}
继续：
...
// todo 消息长度+END_FILE_MIN_BLANK_LENGTH 大于commitLog的空闲空间，则返回END_OF_FILE
if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
this.resetByteBuffer(this.msgStoreItemMemory, maxBlank);
// 1 TOTALSIZE 4字节存储当前文件的剩余空间
this.msgStoreItemMemory.putInt(maxBlank);
// 2 MAGICCODE 4字节存储魔数
this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
// 3 The remaining space may be any value
// Here the length of the specially set maxBlank
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank);
return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(),
queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
}
...
判断剩下的空间能不能放开，如果放不开的话，就塞上一个结束的东西，8个字节是正经的，剩下的随意，然后返回文件满了的状态。
...
// Initialization of storage space
// 初始化存储空间
this.resetByteBuffer(msgStoreItemMemory, msgLen);
// 1 TOTALSIZE
this.msgStoreItemMemory.putInt(msgLen);
// 2 MAGICCODE
this.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);
// 3 BODYCRC
this.msgStoreItemMemory.putInt(msgInner.getBodyCRC());
// 4 QUEUEID
this.msgStoreItemMemory.putInt(msgInner.getQueueId());
// 5 FLAG
this.msgStoreItemMemory.putInt(msgInner.getFlag());
// 6 QUEUEOFFSET
this.msgStoreItemMemory.putLong(queueOffset);
// 7 PHYSICALOFFSET
this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());
// 8 SYSFLAG
this.msgStoreItemMemory.putInt(msgInner.getSysFlag());
// 9 BORNTIMESTAMP
this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
// 10 BORNHOST
this.resetByteBuffer(bornHostHolder, bornHostLength);
this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
// 11 STORETIMESTAMP
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
this.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());
// 15 BODY
this.msgStoreItemMemory.putInt(bodyLength);
if (bodyLength > 0)
this.msgStoreItemMemory.put(msgInner.getBody());
// 16 TOPIC
this.msgStoreItemMemory.put((byte) topicLength);
this.msgStoreItemMemory.put(topicData);
// 17 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);
...
这个就是封装消息了，最后将消息放到byteBuffer中。
...
// 创建AppendMessageResult
AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId,
msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
switch (tranType) {
case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
break;
case MessageSysFlag.TRANSACTION_NOT_TYPE:
case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
// The next update ConsumeQueue information
// 更新消息队列的逻辑偏移量
CommitLog.this.topicQueueTable.put(key, ++queueOffset);
break;
default:
break;
}
return result;
}
最后就是封装追加消息的结果是put_ok，然后更新queue offset ，其实就是+1。
接下来我们回过头来看下appendMessagesInner的后半部分，
...
this.wrotePosition.addAndGet(result.getWroteBytes());
this.storeTimestamp = result.getStoreTimestamp();
return result;
这里其实就是更新了一下 这个MappedFile 写到哪个地方了，更新了下写入时间。
回到commitLog的putMessage方法:
...
// todo 往mappedFile追加消息
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 CompletableFuture.completedFuture(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 CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));
case UNKNOWN_ERROR:
beginTimeInLock = 0;
return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
default:
beginTimeInLock = 0;
return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
}
elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
beginTimeInLock = 0;
} finally {
putMessageLock.unlock();
}
...
这里追加完成了，就需要判断追加状态了，如果是那种MappedFile放不开消息的情况，它会重新获取一个MappedFile，然后重新追加，在释放锁之前，它还会将beginTimeInLock这个字段重置为0；
...
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());
// todo 消息首先进入pagecache，然后执行刷盘操作，
CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);
// todo 接着调用submitReplicaRequest方法将消息提交到HaService,进行数据复制
CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, msg);
// todo 这里使用了ComplateFuture的thenCombine方法，将刷盘、复制当成一
// todo 个联合任务执行，这里设置消息追加的最终状态
return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> {
if (flushStatus != PutMessageStatus.PUT_OK) {
putMessageResult.setPutMessageStatus(flushStatus);
}
if (replicaStatus != PutMessageStatus.PUT_OK) {
putMessageResult.setPutMessageStatus(replicaStatus);
if (replicaStatus == PutMessageStatus.FLUSH_SLAVE_TIMEOUT) {
log.error("do sync transfer other node, wait return, but failed, topic: {} tags: {} client address: {}",
msg.getTopic(), msg.getTags(), msg.getBornHostNameString());
}
}
return putMessageResult;
});
}
判断了一下耗时，如果是大于500ms的话，打印警告，封装put消息的结果，统计store，可以看到后面调用了2个方法，一个是刷盘的，一个是同步消息的，我们这里要看下这个刷盘动作:
public CompletableFuture<PutMessageStatus> submitFlushRequest(AppendMessageResult result, MessageExt messageExt) {
// Synchronization flush 同步刷盘
if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
if (messageExt.isWaitStoreMsgOK()) {
// 构建GroupCommitRequest同步任务并提交到GroupCommitRequest
GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes(),
this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
// 刷盘请求
service.putRequest(request);
return request.future();
} else {
service.wakeup();
return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
}
}
// Asynchronous flush 异步刷盘 这个就是靠os
else {
if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
flushCommitLogService.wakeup();
} else {
commitLogService.wakeup();
}
return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
}
}
如果broker配置的SYNC_FLUSH 并且是个同步消息，这个时候就会创建一个刷盘请求，然后提交刷盘请求，这个时候会等着刷盘完成，默认就是5s。
接着就是到存储器的putMessage方法的后半部分了:
...
// todo 存储消息
CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);
putResultFuture.thenAccept((result) -> {
long elapsedTime = this.getSystemClock().now() - beginTime;
if (elapsedTime > 500) {
log.warn("putMessage not in lock elapsed time(ms)={}, bodyLength={}", elapsedTime, msg.getBody().length);
}
// 记录状态
this.storeStatsService.setPutMessageEntireTimeMax(elapsedTime);
if (null == result || !result.isOk()) {
// 记录状态
this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
}
});
return putResultFuture;
commitlog存入消息之后，咱们这块也就算是完成了，最后就是回到那个processor，然后将put结果写入对应的channel给返回去，告诉消息生产者消息写入结果 。消息存储其实就是找对应的MappedFile，按照一定的格式往文件里面写入，需要注意的是内存映射文件。
这里附一张消息存储字段存储顺序与字段长度的图：
4. 总结
本文主要分析了 broker 接收producer消息的流程，流程如下：
- 处理消息接收的底层服务为 netty，在BrokerController#start方法中启动
- netty服务中，处理消息接收的channelHandler为NettyServerHandler，最终会调用SendMessageProcessor#processRequest来处理消息接收
- 消息接收流程的最后，MappedFile#appendMessage(...)方法会将消息内容写入到commitLog文件中。
网友评论