从本文开始，咱们来剖析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文件中。