从本文开始，我们来分析 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.Sharableclass 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;// 获取最后一个 MappedFileMappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
// 获取写入锁putMessageLock.lock(); //spin or ReentrantLock ,depending on store configtry { 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_FILEif ((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 TOTALSIZEthis.msgStoreItemMemory.putInt(msgLen);// 2 MAGICCODEthis.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);// 3 BODYCRCthis.msgStoreItemMemory.putInt(msgInner.getBodyCRC());// 4 QUEUEIDthis.msgStoreItemMemory.putInt(msgInner.getQueueId());// 5 FLAGthis.msgStoreItemMemory.putInt(msgInner.getFlag());// 6 QUEUEOFFSETthis.msgStoreItemMemory.putLong(queueOffset);// 7 PHYSICALOFFSETthis.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());// 8 SYSFLAGthis.msgStoreItemMemory.putInt(msgInner.getSysFlag());// 9 BORNTIMESTAMPthis.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());// 10 BORNHOSTthis.resetByteBuffer(bornHostHolder, bornHostLength);this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));// 11 STORETIMESTAMPthis.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());// 12 STOREHOSTADDRESSthis.resetByteBuffer(storeHostHolder, storeHostLength);this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));// 13 RECONSUMETIMESthis.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());// 14 Prepared Transaction Offsetthis.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());// 15 BODYthis.msgStoreItemMemory.putInt(bodyLength);if (bodyLength > 0) this.msgStoreItemMemory.put(msgInner.getBody());// 16 TOPICthis.msgStoreItemMemory.put((byte) topicLength);this.msgStoreItemMemory.put(topicData);// 17 PROPERTIESthis.msgStoreItemMemory.putShort((short) propertiesLength);if (propertiesLength > 0) this.msgStoreItemMemory.put(propertiesData);
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();// Write messages to the queue bufferbyteBuffer.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 文件中。
评论