RocketMQ中Broker的訊息儲存原始碼分析
阿新 • • 發佈：2019-08-06
Broker和前面分析過的NameServer類似，需要在Pipeline責任鏈上通過NettyServerHandler來處理訊息
【RocketMQ中NameServer的啟動原始碼分析】
實際上就通過前面提到的SendMessageProcessor的processRequest方法處理
【RocketMQ中Broker的啟動原始碼分析（一）】
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，詳見
【RocketMQ中Producer訊息的傳送原始碼分析】
之後會對訊息的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），以及主從複製
&n