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），以及主從複製
