RocketMQ中Broker的消息存储源码分析
- 2019 年 10 月 3 日
- 筆記
Broker和前面分析过的NameServer类似，需要在Pipeline责任链上通过NettyServerHandler来处理消息
实际上就通过前面提到的SendMessageProcessor的processRequest方法处理
SendMessageProcessor的processRequest方法：
1 public RemotingCommand processRequest(ChannelHandlerContext ctx, 2 RemotingCommand request) throws RemotingCommandException { 3 SendMessageContext mqtraceContext; 4 switch (request.getCode()) { 5 case RequestCode.CONSUMER_SEND_MSG_BACK: 6 return this.consumerSendMsgBack(ctx, request); 7 default: 8 SendMessageRequestHeader requestHeader = parseRequestHeader(request); 9 if (requestHeader == null) { 10 return null; 11 } 12 13 mqtraceContext = buildMsgContext(ctx, requestHeader); 14 this.executeSendMessageHookBefore(ctx, request, mqtraceContext); 15 16 RemotingCommand response; 17 if (requestHeader.isBatch()) { 18 response = this.sendBatchMessage(ctx, request, mqtraceContext, requestHeader); 19 } else { 20 response = this.sendMessage(ctx, request, mqtraceContext, requestHeader); 21 } 22 23 this.executeSendMessageHookAfter(response, mqtraceContext); 24 return response; 25 } 26 }
这里讨论Producer发送的消息，直接进入default语句
根据请求RemotingCommand，通过parseRequestHeader以及buildMsgContext方法，解析RemotingCommand中的相应信息，再封装到SendMessageRequestHeader和SendMessageContext中
接着调用executeSendMessageHookBefore方法：
1 public void executeSendMessageHookBefore(final ChannelHandlerContext ctx, final RemotingCommand request, 2 SendMessageContext context) { 3 if (hasSendMessageHook()) { 4 for (SendMessageHook hook : this.sendMessageHookList) { 5 try { 6 final SendMessageRequestHeader requestHeader = parseRequestHeader(request); 7 8 if (null != requestHeader) { 9 context.setProducerGroup(requestHeader.getProducerGroup()); 10 context.setTopic(requestHeader.getTopic()); 11 context.setBodyLength(request.getBody().length); 12 context.setMsgProps(requestHeader.getProperties()); 13 context.setBornHost(RemotingHelper.parseChannelRemoteAddr(ctx.channel())); 14 context.setBrokerAddr(this.brokerController.getBrokerAddr()); 15 context.setQueueId(requestHeader.getQueueId()); 16 } 17 18 hook.sendMessageBefore(context); 19 if (requestHeader != null) { 20 requestHeader.setProperties(context.getMsgProps()); 21 } 22 } catch (Throwable e) { 23 // Ignore 24 } 25 } 26 } 27 }
这里会执行所有SendMessageHook钩子的sendMessageBefore方法
然后调用sendMessage方法。进一步处理
sendMessage方法：
1 private RemotingCommand sendMessage(final ChannelHandlerContext ctx, 2 final RemotingCommand request, 3 final SendMessageContext sendMessageContext, 4 final SendMessageRequestHeader requestHeader) throws RemotingCommandException { 5 6 final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class); 7 final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader)response.readCustomHeader(); 8 9 response.setOpaque(request.getOpaque()); 10 11 response.addExtField(MessageConst.PROPERTY_MSG_REGION, this.brokerController.getBrokerConfig().getRegionId()); 12 response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(this.brokerController.getBrokerConfig().isTraceOn())); 13 14 log.debug("receive SendMessage request command, {}", request); 15 16 final long startTimstamp = this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp(); 17 if (this.brokerController.getMessageStore().now() < startTimstamp) { 18 response.setCode(ResponseCode.SYSTEM_ERROR); 19 response.setRemark(String.format("broker unable to service, until %s", UtilAll.timeMillisToHumanString2(startTimstamp))); 20 return response; 21 } 22 23 response.setCode(-1); 24 super.msgCheck(ctx, requestHeader, response); 25 if (response.getCode() != -1) { 26 return response; 27 } 28 29 final byte[] body = request.getBody(); 30 31 int queueIdInt = requestHeader.getQueueId(); 32 TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic()); 33 34 if (queueIdInt < 0) { 35 queueIdInt = Math.abs(this.random.nextInt() % 99999999) % topicConfig.getWriteQueueNums(); 36 } 37 38 MessageExtBrokerInner msgInner = new MessageExtBrokerInner(); 39 msgInner.setTopic(requestHeader.getTopic()); 40 msgInner.setQueueId(queueIdInt); 41 42 if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig)) { 43 return response; 44 } 45 46 msgInner.setBody(body); 47 msgInner.setFlag(requestHeader.getFlag()); 48 MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties())); 49 msgInner.setPropertiesString(requestHeader.getProperties()); 50 msgInner.setBornTimestamp(requestHeader.getBornTimestamp()); 51 msgInner.setBornHost(ctx.channel().remoteAddress()); 52 msgInner.setStoreHost(this.getStoreHost()); 53 msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes()); 54 PutMessageResult putMessageResult = null; 55 Map<String, String> oriProps = MessageDecoder.string2messageProperties(requestHeader.getProperties()); 56 String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED); 57 if (traFlag != null && Boolean.parseBoolean(traFlag)) { 58 if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) { 59 response.setCode(ResponseCode.NO_PERMISSION); 60 response.setRemark( 61 "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1() 62 + "] sending transaction message is forbidden"); 63 return response; 64 } 65 putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner); 66 } else { 67 putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner); 68 } 69 70 return handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt); 71 72 }
这里首先会把具体的消息及其相关信息封装在MessageExtBrokerInner中
MessageExtBrokerInner继承自Message，详见
之后会对消息的PROPERTY_TRANSACTION_PREPARED属性进行检查，判断是否是事务消息
若是事务消息，会检查是否设置了拒绝事务消息的配置rejectTransactionMessage
若是拒绝则返回相应响应response，由Netty发送给Producer
否则调用TransactionalMessageService的prepareMessage方法
若不是事务消息则调用MessageStore的putMessage方法
在事务消息的处理里，实际上只是对MessageExtBrokerInner设置相应的属性，最后还是调用putMessage方法：
1 public PutMessageResult prepareMessage(MessageExtBrokerInner messageInner) { 2 return transactionalMessageBridge.putHalfMessage(messageInner); 3 } 4 5 public PutMessageResult putHalfMessage(MessageExtBrokerInner messageInner) { 6 return store.putMessage(parseHalfMessageInner(messageInner)); 7 } 8 9 private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) { 10 MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic()); 11 MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID, 12 String.valueOf(msgInner.getQueueId())); 13 msgInner.setSysFlag( 14 MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE)); 15 msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic()); 16 msgInner.setQueueId(0); 17 msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties())); 18 return msgInner; 19 }
DefaultMessageStore的putMessage方法：
1 public PutMessageResult putMessage(MessageExtBrokerInner msg) { 2 if (this.shutdown) { 3 log.warn("message store has shutdown, so putMessage is forbidden"); 4 return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null); 5 } 6 7 if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) { 8 long value = this.printTimes.getAndIncrement(); 9 if ((value % 50000) == 0) { 10 log.warn("message store is slave mode, so putMessage is forbidden "); 11 } 12 13 return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null); 14 } 15 16 if (!this.runningFlags.isWriteable()) { 17 long value = this.printTimes.getAndIncrement(); 18 if ((value % 50000) == 0) { 19 log.warn("message store is not writeable, so putMessage is forbidden " + this.runningFlags.getFlagBits()); 20 } 21 22 return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null); 23 } else { 24 this.printTimes.set(0); 25 } 26 27 if (msg.getTopic().length() > Byte.MAX_VALUE) { 28 log.warn("putMessage message topic length too long " + msg.getTopic().length()); 29 return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null); 30 } 31 32 if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) { 33 log.warn("putMessage message properties length too long " + msg.getPropertiesString().length()); 34 return new PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, null); 35 } 36 37 if (this.isOSPageCacheBusy()) { 38 return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, null); 39 } 40 41 long beginTime = this.getSystemClock().now(); 42 PutMessageResult result = this.commitLog.putMessage(msg); 43 44 long eclipseTime = this.getSystemClock().now() - beginTime; 45 if (eclipseTime > 500) { 46 log.warn("putMessage not in lock eclipse time(ms)={}, bodyLength={}", eclipseTime, msg.getBody().length); 47 } 48 this.storeStatsService.setPutMessageEntireTimeMax(eclipseTime); 49 50 if (null == result || !result.isOk()) { 51 this.storeStatsService.getPutMessageFailedTimes().incrementAndGet(); 52 } 53 54 return result; 55 }
这里会对消息的合法性以及Broker的状态做一系列的检查，在全部通过后才继续，否则返回带有相应提示的响应
其中会检查Broker是否是SLAVE
若是SLAVE，会返回SERVICE_NOT_AVAILABLE，不允许Slave直接存储来自Producer的消息，间接说明了Master和Slave的主从关系
满足所有条件后，调用commitLog的putMessage方法：
1 public PutMessageResult putMessage(final MessageExtBrokerInner msg) { 2 // Set the storage time 3 msg.setStoreTimestamp(System.currentTimeMillis()); 4 // Set the message body BODY CRC (consider the most appropriate setting 5 // on the client) 6 msg.setBodyCRC(UtilAll.crc32(msg.getBody())); 7 // Back to Results 8 AppendMessageResult result = null; 9 10 StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService(); 11 12 String topic = msg.getTopic(); 13 int queueId = msg.getQueueId(); 14 15 final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag()); 16 if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE 17 || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) { 18 // Delay Delivery 19 if (msg.getDelayTimeLevel() > 0) { 20 if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) { 21 msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()); 22 } 23 24 topic = ScheduleMessageService.SCHEDULE_TOPIC; 25 queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel()); 26 27 // Backup real topic, queueId 28 MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic()); 29 MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId())); 30 msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties())); 31 32 msg.setTopic(topic); 33 msg.setQueueId(queueId); 34 } 35 } 36 37 long eclipseTimeInLock = 0; 38 MappedFile unlockMappedFile = null; 39 MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); 40 41 putMessageLock.lock(); //spin or ReentrantLock ,depending on store config 42 try { 43 long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now(); 44 this.beginTimeInLock = beginLockTimestamp; 45 46 // Here settings are stored timestamp, in order to ensure an orderly 47 // global 48 msg.setStoreTimestamp(beginLockTimestamp); 49 50 if (null == mappedFile || mappedFile.isFull()) { 51 mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise 52 } 53 if (null == mappedFile) { 54 log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString()); 55 beginTimeInLock = 0; 56 return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null); 57 } 58 59 result = mappedFile.appendMessage(msg, this.appendMessageCallback); 60 switch (result.getStatus()) { 61 case PUT_OK: 62 break; 63 case END_OF_FILE: 64 unlockMappedFile = mappedFile; 65 // Create a new file, re-write the message 66 mappedFile = this.mappedFileQueue.getLastMappedFile(0); 67 if (null == mappedFile) { 68 // XXX: warn and notify me 69 log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString()); 70 beginTimeInLock = 0; 71 return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result); 72 } 73 result = mappedFile.appendMessage(msg, this.appendMessageCallback); 74 break; 75 case MESSAGE_SIZE_EXCEEDED: 76 case PROPERTIES_SIZE_EXCEEDED: 77 beginTimeInLock = 0; 78 return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result); 79 case UNKNOWN_ERROR: 80 beginTimeInLock = 0; 81 return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result); 82 default: 83 beginTimeInLock = 0; 84 return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result); 85 } 86 87 eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp; 88 beginTimeInLock = 0; 89 } finally { 90 putMessageLock.unlock(); 91 } 92 93 if (eclipseTimeInLock > 500) { 94 log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipseTimeInLock, msg.getBody().length, result); 95 } 96 97 if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) { 98 this.defaultMessageStore.unlockMappedFile(unlockMappedFile); 99 } 100 101 PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result); 102 103 // Statistics 104 storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet(); 105 storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes()); 106 107 handleDiskFlush(result, putMessageResult, msg); 108 handleHA(result, putMessageResult, msg); 109 110 return putMessageResult; 111 }
这里会通过mappedFileQueue的getLastMappedFile方法，找到CommitLog文件对应的映射MappedFile
关于MappedFile，及其一些操作，在 【RocketMQ中Broker的启动源码分析（二）】 中关于消息的调度时分析过了，这里涉及到就不再累赘
然后调用MappedFile的appendMessage方法，其中参数appendMessageCallback，是在CommitLog构造时设置的，其是实现类是CommitLog的内部类，用于后面appendMessage操作的回调在CommitLog中进行
MappedFile的appendMessage方法：
1 public AppendMessageResult appendMessage(final MessageExtBrokerInner msg, final AppendMessageCallback cb) { 2 return appendMessagesInner(msg, cb); 3 } 4 5 public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb) { 6 assert messageExt != null; 7 assert cb != null; 8 9 int currentPos = this.wrotePosition.get(); 10 11 if (currentPos < this.fileSize) { 12 ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice(); 13 byteBuffer.position(currentPos); 14 AppendMessageResult result = null; 15 if (messageExt instanceof MessageExtBrokerInner) { 16 result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt); 17 } else if (messageExt instanceof MessageExtBatch) { 18 result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt); 19 } else { 20 return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR); 21 } 22 this.wrotePosition.addAndGet(result.getWroteBytes()); 23 this.storeTimestamp = result.getStoreTimestamp(); 24 return result; 25 } 26 log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize); 27 return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR); 28 }
在这里要注意这一步：
1 ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
可以得到一个共享的子缓冲区ByteBuffer
稍微提一下，只有当Broker使用异步刷盘并且开启内存字节缓冲区的情况下，writeBuffer才有意义，否则都是mappedByteBuffer
后续再介绍
然后调用刚才设置的回调接口的doAppend方法：
1 public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank, 2 final MessageExtBrokerInner msgInner) { 3 // STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br> 4 5 // PHY OFFSET 6 long wroteOffset = fileFromOffset + byteBuffer.position(); 7 8 this.resetByteBuffer(hostHolder, 8); 9 String msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(hostHolder), wroteOffset); 10 11 // Record ConsumeQueue information 12 keyBuilder.setLength(0); 13 keyBuilder.append(msgInner.getTopic()); 14 keyBuilder.append('-'); 15 keyBuilder.append(msgInner.getQueueId()); 16 String key = keyBuilder.toString(); 17 Long queueOffset = CommitLog.this.topicQueueTable.get(key); 18 if (null == queueOffset) { 19 queueOffset = 0L; 20 CommitLog.this.topicQueueTable.put(key, queueOffset); 21 } 22 23 // Transaction messages that require special handling 24 final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag()); 25 switch (tranType) { 26 // Prepared and Rollback message is not consumed, will not enter the 27 // consumer queuec 28 case MessageSysFlag.TRANSACTION_PREPARED_TYPE: 29 case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE: 30 queueOffset = 0L; 31 break; 32 case MessageSysFlag.TRANSACTION_NOT_TYPE: 33 case MessageSysFlag.TRANSACTION_COMMIT_TYPE: 34 default: 35 break; 36 } 37 38 /** 39 * Serialize message 40 */ 41 final byte[] propertiesData = 42 msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8); 43 44 final int propertiesLength = propertiesData == null ? 0 : propertiesData.length; 45 46 if (propertiesLength > Short.MAX_VALUE) { 47 log.warn("putMessage message properties length too long. length={}", propertiesData.length); 48 return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED); 49 } 50 51 final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8); 52 final int topicLength = topicData.length; 53 54 final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length; 55 56 final int msgLen = calMsgLength(bodyLength, topicLength, propertiesLength); 57 58 // Exceeds the maximum message 59 if (msgLen > this.maxMessageSize) { 60 CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLength 61 + ", maxMessageSize: " + this.maxMessageSize); 62 return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED); 63 } 64 65 // Determines whether there is sufficient free space 66 if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) { 67 this.resetByteBuffer(this.msgStoreItemMemory, maxBlank); 68 // 1 TOTALSIZE 69 this.msgStoreItemMemory.putInt(maxBlank); 70 // 2 MAGICCODE 71 this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE); 72 // 3 The remaining space may be any value 73 // Here the length of the specially set maxBlank 74 final long beginTimeMills = CommitLog.this.defaultMessageStore.now(); 75 byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank); 76 return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(), 77 queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills); 78 } 79 80 // Initialization of storage space 81 this.resetByteBuffer(msgStoreItemMemory, msgLen); 82 // 1 TOTALSIZE 83 this.msgStoreItemMemory.putInt(msgLen); 84 // 2 MAGICCODE 85 this.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE); 86 // 3 BODYCRC 87 this.msgStoreItemMemory.putInt(msgInner.getBodyCRC()); 88 // 4 QUEUEID 89 this.msgStoreItemMemory.putInt(msgInner.getQueueId()); 90 // 5 FLAG 91 this.msgStoreItemMemory.putInt(msgInner.getFlag()); 92 // 6 QUEUEOFFSET 93 this.msgStoreItemMemory.putLong(queueOffset); 94 // 7 PHYSICALOFFSET 95 this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position()); 96 // 8 SYSFLAG 97 this.msgStoreItemMemory.putInt(msgInner.getSysFlag()); 98 // 9 BORNTIMESTAMP 99 this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp()); 100 // 10 BORNHOST 101 this.resetByteBuffer(hostHolder, 8); 102 this.msgStoreItemMemory.put(msgInner.getBornHostBytes(hostHolder)); 103 // 11 STORETIMESTAMP 104 this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp()); 105 // 12 STOREHOSTADDRESS 106 this.resetByteBuffer(hostHolder, 8); 107 this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(hostHolder)); 108 //this.msgBatchMemory.put(msgInner.getStoreHostBytes()); 109 // 13 RECONSUMETIMES 110 this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes()); 111 // 14 Prepared Transaction Offset 112 this.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset()); 113 // 15 BODY 114 this.msgStoreItemMemory.putInt(bodyLength); 115 if (bodyLength > 0) 116 this.msgStoreItemMemory.put(msgInner.getBody()); 117 // 16 TOPIC 118 this.msgStoreItemMemory.put((byte) topicLength); 119 this.msgStoreItemMemory.put(topicData); 120 // 17 PROPERTIES 121 this.msgStoreItemMemory.putShort((short) propertiesLength); 122 if (propertiesLength > 0) 123 this.msgStoreItemMemory.put(propertiesData); 124 125 final long beginTimeMills = CommitLog.this.defaultMessageStore.now(); 126 // Write messages to the queue buffer 127 byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen); 128 129 AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId, 130 msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills); 131 132 switch (tranType) { 133 case MessageSysFlag.TRANSACTION_PREPARED_TYPE: 134 case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE: 135 break; 136 case MessageSysFlag.TRANSACTION_NOT_TYPE: 137 case MessageSysFlag.TRANSACTION_COMMIT_TYPE: 138 // The next update ConsumeQueue information 139 CommitLog.this.topicQueueTable.put(key, ++queueOffset); 140 break; 141 default: 142 break; 143 } 144 return result; 145 }
这里首先会根据fileFromOffset和byteBuffer的position计算出实际要往文件写入时的Offset，使用wroteOffset记录
之后根据MessageExtBrokerInner中的内容，按照CommitLog文件的消息结构，通过put操作将消息缓存在msgStoreItemMemory这个ByteBuffer中
CommitLog文件中消息的结构：
然后通过：
1 byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);
将msgStoreItemMemory中的信息缓存到刚才获取的这个共享ByteBuffer中
其中：
1 if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) { 2 this.resetByteBuffer(this.msgStoreItemMemory, maxBlank); 3 // 1 TOTALSIZE 4 this.msgStoreItemMemory.putInt(maxBlank); 5 // 2 MAGICCODE 6 this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE); 7 // 3 The remaining space may be any value 8 // Here the length of the specially set maxBlank 9 final long beginTimeMills = CommitLog.this.defaultMessageStore.now(); 10 byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank); 11 return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(), 12 queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills); 13 }
会检查当前CommitLog文件是否有可用空间，CommitLog结尾会以BLANK（8字节）的形式出现
这里可以看到BLANK的结构，由4字节maxBlank（fileSize-currentPos）以及4字节魔数构成
文件空间不可用会返回END_OF_FILE状态信息，之后会有用
回到appendMessagesInner方法，在完成doAppend后，根据往缓冲区写入的数据大小，修改wrotePosition这个AtomicInteger值，以便下次的定位
再回到CommitLog的putMessage方法：
1 result = mappedFile.appendMessage(msg, this.appendMessageCallback); 2 switch (result.getStatus()) { 3 case PUT_OK: 4 break; 5 case END_OF_FILE: 6 unlockMappedFile = mappedFile; 7 // Create a new file, re-write the message 8 mappedFile = this.mappedFileQueue.getLastMappedFile(0); 9 if (null == mappedFile) { 10 // XXX: warn and notify me 11 log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString()); 12 beginTimeInLock = 0; 13 return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result); 14 } 15 result = mappedFile.appendMessage(msg, this.appendMessageCallback); 16 break; 17 case MESSAGE_SIZE_EXCEEDED: 18 case PROPERTIES_SIZE_EXCEEDED: 19 beginTimeInLock = 0; 20 return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result); 21 case UNKNOWN_ERROR: 22 beginTimeInLock = 0; 23 return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result); 24 default: 25 beginTimeInLock = 0; 26 return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result); 27 }
在得到result后，会进行状态检查
其中若是刚才说都END_OF_FILE状态
则会通过mappedFileQueue的getLastMappedFile方法，创建一个新的CommitLog文件以及文件映射MappedFile
然后调用这个新的MappedFile的appendMessage方法，重复之前的步骤，这样就会将消息往新的ByteBuffer中，而之前的那个则缓存着8字节的BLANK
到这里SendMessageProcessor的任务其实已经完成的差不多了，但是，按照我上面的分析来看，仅仅只是将消息进行了缓存，并没有真正地写入磁盘完成持久化
在CommitLog的putMessage方法最后还有两步非常重要的操作：
1 handleDiskFlush(result, putMessageResult, msg); 2 handleHA(result, putMessageResult, msg);
handleHA在后续分析主从复制时再说
那么重点是这个handleDiskFlush方法：
1 public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) { 2 // Synchronization flush 3 if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) { 4 final GroupCommitService service = (GroupCommitService) this.flushCommitLogService; 5 if (messageExt.isWaitStoreMsgOK()) { 6 GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes()); 7 service.putRequest(request); 8 boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout()); 9 if (!flushOK) { 10 log.error("do groupcommit, wait for flush failed, topic: " + messageExt.getTopic() + " tags: " + messageExt.getTags() 11 + " client address: " + messageExt.getBornHostString()); 12 putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT); 13 } 14 } else { 15 service.wakeup(); 16 } 17 } 18 // Asynchronous flush 19 else { 20 if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) { 21 flushCommitLogService.wakeup(); 22 } else { 23 commitLogService.wakeup(); 24 } 25 } 26 }
这里的话就会涉及到Broker的CommitLog刷盘
关于CommitLog刷盘我会在下一篇博客详细分析，这里我就简单说一下
Broker的CommitLog刷盘会启动一个线程，不停地将缓冲区的内容写入磁盘（CommitLog文件）中，主要分为异步刷盘和同步刷盘
异步刷盘又可以分为两种方式：
①缓存到mappedByteBuffer -> 写入磁盘（包括同步刷盘）
②缓存到writeBuffer -> 缓存到fileChannel -> 写入磁盘 （前面说过的开启内存字节缓冲区情况下）
也就是说Broker在接收到Producer的消息时，并没有同时将消息持久化，而是进行缓存记录，然后通过刷盘线程，将缓存写入磁盘完成持久化
在上一篇博客我还详细分析过消息调度
结合着来看，在刷盘线程工作的同时，调度线程也在从磁盘读取消息到内存，将消息进行分配，刷盘线程也会随时写入新消息，二者相互协调
RocketMQ的工作原理到这已经初见端倪，后续重点会分析消费者的消费（Pull、Push），以及主从复制
