RocketMQ源碼解析：Message存儲

時間 2019-11-17

標籤 rocketmq 源碼解析 message 存儲简体版

原文原文鏈接

🙂🙂🙂關注微信公衆號：【芋艿的後端小屋】有福利： node

RocketMQ / MyCAT / Sharding-JDBC 全部源碼分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文註釋源碼 GitHub 地址

您對於源碼的疑問每條留言都將獲得認真回覆。甚至不知道如何讀源碼也能夠請教噢。

新的源碼解析文章實時收到通知。每週更新一篇左右。

一、概述

本文接《RocketMQ 源碼分析 —— Message 發送與接收》。
主要解析 CommitLog 存儲消息部分。後端

二、CommitLog 結構

CommitLog、MappedFileQueue、MappedFile 的關係以下：bash

CommitLog、MappedFileQueue、MappedFile的關係

CommitLog : MappedFileQueue : MappedFile = 1 : 1 : N。

反應到系統文件以下：微信

Yunai-MacdeMacBook-Pro-2:commitlog yunai$ pwd
/Users/yunai/store/commitlog
Yunai-MacdeMacBook-Pro-2:commitlog yunai$ ls -l
total 10485760
-rw-r--r--  1 yunai  staff  1073741824  4 21 16:27 00000000000000000000
-rw-r--r--  1 yunai  staff  1073741824  4 21 16:29 00000000001073741824
-rw-r--r--  1 yunai  staff  1073741824  4 21 16:32 00000000002147483648
-rw-r--r--  1 yunai  staff  1073741824  4 21 16:33 00000000003221225472
-rw-r--r--  1 yunai  staff  1073741824  4 21 16:32 00000000004294967296複製代碼

CommitLog、MappedFileQueue、MappedFile 的定義以下：架構

MappedFile ：00000000000000000000、0000000000107374182四、00000000002147483648等文件。
MappedFileQueue ：MappedFile 所在的文件夾，對 MappedFile 進行封裝成文件隊列，對上層提供可無限使用的文件容量。
- 每一個 MappedFile 統一文件大小。
- 文件命名方式：fileName[n] = fileName[n - 1] + mappedFileSize。在 CommitLog 裏默認爲 1GB。
CommitLog ：針對 MappedFileQueue 的封裝使用。

CommitLog 目前存儲在 MappedFile 有兩種內容類型：app

MESSAGE ：消息。
BLANK ：文件不足以存儲消息時的空白佔位。

CommitLog 存儲在 MappedFile的結構：eclipse

MESSAGE[1] MESSAGE[2] ... MESSAGE[n - 1] MESSAGE[n] BLANK

MESSAGE 在 CommitLog 存儲結構：異步

第幾位	字段	說明	數據類型	字節數
1	MsgLen	消息總長度	Int	4
2	MagicCode	MESSAGE_MAGIC_CODE	Int	4
3	BodyCRC	消息內容CRC	Int	4
4	QueueId	消息隊列編號	Int	4
5	Flag	flag	Int	4
6	QueueOffset	消息隊列位置	Long	8
7	PhysicalOffset	物理位置。在 `CommitLog` 的順序存儲位置。	Long	8
8	SysFlag	MessageSysFlag	Int	4
9	BornTimestamp	生成消息時間戳	Long	8
10	BornHost	生效消息的地址+端口	Long	8
11	StoreTimestamp	存儲消息時間戳	Long	8
12	StoreHost	存儲消息的地址+端口	Long	8
13	ReconsumeTimes	從新消費消息次數	Int	4
14	PreparedTransationOffset		Long	8
15	BodyLength + Body	內容長度 + 內容	Int + Bytes	4 + bodyLength
16	TopicLength + Topic	Topic長度 + Topic	Byte + Bytes	1 + topicLength
17	PropertiesLength + Properties	拓展字段長度 + 拓展字段	Short + Bytes	2 + PropertiesLength

BLANK 在 CommitLog 存儲結構：ide

第幾位	字段	說明	數據類型	字節數
1	maxBlank	空白長度	Int	4
2	MagicCode	BLANK_MAGIC_CODE	Int	4

三、CommitLog 存儲消息

Broker存儲發送消息順序圖

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:     // 事務相關 TODO 待讀：事務相關
 16:     final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
 17:     if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE//
 18:         || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
 19:         // Delay Delivery
 20:         if (msg.getDelayTimeLevel() > 0) {
 21:             if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
 22:                 msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
 23:             }
 24: 
 25:             topic = ScheduleMessageService.SCHEDULE_TOPIC;
 26:             queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
 27: 
 28:             // Backup real topic, queueId
 29:             MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
 30:             MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
 31:             msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
 32: 
 33:             msg.setTopic(topic);
 34:             msg.setQueueId(queueId);
 35:         }
 36:     }
 37: 
 38:     long eclipseTimeInLock = 0;
 39: 
 40:     // 獲取寫入映射文件
 41:     MappedFile unlockMappedFile = null;
 42:     MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
 43: 
 44:     // 獲取寫入鎖
 45:     lockForPutMessage(); //spin...
 46:     try {
 47:         long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
 48:         this.beginTimeInLock = beginLockTimestamp;
 49: 
 50:         // Here settings are stored timestamp, in order to ensure an orderly
 51:         // global
 52:         msg.setStoreTimestamp(beginLockTimestamp);
 53: 
 54:         // 當不存在映射文件時，進行建立
 55:         if (null == mappedFile || mappedFile.isFull()) {
 56:             mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
 57:         }
 58:         if (null == mappedFile) {
 59:             log.error("create maped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
 60:             beginTimeInLock = 0;
 61:             return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
 62:         }
 63: 
 64:         // 存儲消息
 65:         result = mappedFile.appendMessage(msg, this.appendMessageCallback);
 66:         switch (result.getStatus()) {
 67:             case PUT_OK:
 68:                 break;
 69:             case END_OF_FILE: // 當文件尾時，獲取新的映射文件，並進行插入
 70:                 unlockMappedFile = mappedFile;
 71:                 // Create a new file, re-write the message
 72:                 mappedFile = this.mappedFileQueue.getLastMappedFile(0);
 73:                 if (null == mappedFile) {
 74:                     // XXX: warn and notify me
 75:                     log.error("create maped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
 76:                     beginTimeInLock = 0;
 77:                     return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result);
 78:                 }
 79:                 result = mappedFile.appendMessage(msg, this.appendMessageCallback);
 80:                 break;
 81:             case MESSAGE_SIZE_EXCEEDED:
 82:             case PROPERTIES_SIZE_EXCEEDED:
 83:                 beginTimeInLock = 0;
 84:                 return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result);
 85:             case UNKNOWN_ERROR:
 86:                 beginTimeInLock = 0;
 87:                 return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
 88:             default:
 89:                 beginTimeInLock = 0;
 90:                 return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
 91:         }
 92: 
 93:         eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
 94:         beginTimeInLock = 0;
 95:     } finally {
 96:         // 釋放寫入鎖
 97:         releasePutMessageLock();
 98:     }
 99: 
100:     if (eclipseTimeInLock > 500) {
101:         log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipseTimeInLock, msg.getBody().length, result);
102:     }
103: 
104:     // 
105:     if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
106:         this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
107:     }
108: 
109:     PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
110: 
111:     // Statistics
112:     storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
113:     storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
114: 
115:     // 進行同步||異步 flush||commit
116:     GroupCommitRequest request = null;
117:     // Synchronization flush
118:     if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
119:         final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
120:         if (msg.isWaitStoreMsgOK()) {
121:             request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
122:             service.putRequest(request);
123:             boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
124:             if (!flushOK) {
125:                 log.error("do groupcommit, wait for flush failed, topic: " + msg.getTopic() + " tags: " + msg.getTags()
126:                     + " client address: " + msg.getBornHostString());
127:                 putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
128:             }
129:         } else {
130:             service.wakeup();
131:         }
132:     }
133:     // Asynchronous flush
134:     else {
135:         if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
136:             flushCommitLogService.wakeup(); // important：喚醒commitLog線程，進行flush
137:         } else {
138:             commitLogService.wakeup();
139:         }
140:     }
141: 
142:     // Synchronous write double 若是是同步Master，同步到從節點 // TODO 待讀：數據同步
143:     if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
144:         HAService service = this.defaultMessageStore.getHaService();
145:         if (msg.isWaitStoreMsgOK()) {
146:             // Determine whether to wait
147:             if (service.isSlaveOK(result.getWroteOffset() + result.getWroteBytes())) {
148:                 if (null == request) {
149:                     request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
150:                 }
151:                 service.putRequest(request);
152: 
153:                 service.getWaitNotifyObject().wakeupAll();
154: 
155:                 boolean flushOK =
156:                     // TODO
157:                     request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
158:                 if (!flushOK) {
159:                     log.error("do sync transfer other node, wait return, but failed, topic: " + msg.getTopic() + " tags: "
160:                         + msg.getTags() + " client address: " + msg.getBornHostString());
161:                     putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
162:                 }
163:             }
164:             // Slave problem
165:             else {
166:                 // Tell the producer, slave not available
167:                 putMessageResult.setPutMessageStatus(PutMessageStatus.SLAVE_NOT_AVAILABLE);
168:             }
169:         }
170:     }
171: 
172:     return putMessageResult;
173: }複製代碼

說明：存儲消息，並返回存儲結果。
第 2 行：設置存儲時間等。
第 16 至 36 行：事務消息相關，暫未了解。
第 45 & 97 行：獲取鎖與釋放鎖。
第 52 行：再次設置存儲時間。目前會有多處地方設置存儲時間。
第 55 至 62 行：獲取 MappedFile，若不存在或已滿，則進行建立。詳細解析見：MappedFileQueue#getLastMappedFile(...)。
第 65 行：插入消息到 MappedFile，解析解析見：MappedFile#appendMessage(...)。
第 69 至 80 行：MappedFile 已滿，建立新的，再次插入消息。
第 116 至 140 行：消息刷盤，即持久化到文件。上面插入消息實際未存儲到硬盤。此處，根據不一樣的刷盤策略，執行會有不一樣。詳細解析見：FlushCommitLogService。
第 143 至 173 行：Broker 主從同步。後面的文章會詳細解析😈。

MappedFileQueue#getLastMappedFile(...)

1: public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {
  2:     long createOffset = -1; // 建立文件開始offset。-1時，不建立
  3:     MappedFile mappedFileLast = getLastMappedFile();
  4: 
  5:     if (mappedFileLast == null) { // 一個映射文件都不存在
  6:         createOffset = startOffset - (startOffset % this.mappedFileSize);
  7:     }
  8: 
  9:     if (mappedFileLast != null && mappedFileLast.isFull()) { // 最後一個文件已滿
 10:         createOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;
 11:     }
 12: 
 13:     if (createOffset != -1 && needCreate) { // 建立文件
 14:         String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);
 15:         String nextNextFilePath = this.storePath + File.separator
 16:             + UtilAll.offset2FileName(createOffset + this.mappedFileSize);
 17:         MappedFile mappedFile = null;
 18: 
 19:         if (this.allocateMappedFileService != null) {
 20:             mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath,
 21:                 nextNextFilePath, this.mappedFileSize);
 22:         } else {
 23:             try {
 24:                 mappedFile = new MappedFile(nextFilePath, this.mappedFileSize);
 25:             } catch (IOException e) {
 26:                 log.error("create mappedFile exception", e);
 27:             }
 28:         }
 29: 
 30:         if (mappedFile != null) {
 31:             if (this.mappedFiles.isEmpty()) {
 32:                 mappedFile.setFirstCreateInQueue(true);
 33:             }
 34:             this.mappedFiles.add(mappedFile);
 35:         }
 36: 
 37:         return mappedFile;
 38:     }
 39: 
 40:     return mappedFileLast;
 41: }複製代碼

說明：獲取最後一個 MappedFile，若不存在或文件已滿，則進行建立。
第 5 至 11 行：計算當文件不存在或已滿時，新建立文件的 createOffset。
第 14 行：計算文件名。今後處咱們可
以得知，MappedFile的文件命名規則：源碼分析

fileName[n] = fileName[n - 1] + n * mappedFileSize
fileName[0] = startOffset - (startOffset % this.mappedFileSize)

目前 CommitLog 的 startOffset 爲 0。
此處有個疑問，爲何須要 (startOffset % this.mappedFileSize)。例如：

| startOffset | mappedFileSize | createOffset |
| --- | :-- | :-- |
| 5 | 1 | 5 |
| 5 | 2 | 4 |
| 5 | 3 | 3 |
| 5 | 4 | 4 |
| 5 | > 5 | 0 |

若是有知道的同窗，麻煩提示下。😈
解答：fileName[0] = startOffset - (startOffset % this.mappedFileSize) 計算出來的是，以 this.mappedFileSize 爲每一個文件大小時，startOffset 所在文件的開始offset
第 30 至 35 行：設置 MappedFile是不是第一個建立的文件。該標識用於 ConsumeQueue 對應的 MappedFile ，詳見 ConsumeQueue#fillPreBlank。

MappedFile#appendMessage(...)

1: public AppendMessageResult appendMessage(final MessageExtBrokerInner msg, final AppendMessageCallback cb) {
  2:     assert msg != null;
  3:     assert cb != null;
  4: 
  5:     int currentPos = this.wrotePosition.get();
  6: 
  7:     if (currentPos < this.fileSize) {
  8:         ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
  9:         byteBuffer.position(currentPos);
 10:         AppendMessageResult result =
 11:             cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, msg);
 12:         this.wrotePosition.addAndGet(result.getWroteBytes());
 13:         this.storeTimestamp = result.getStoreTimestamp();
 14:         return result;
 15:     }
 16: 
 17:     log.error("MappedFile.appendMessage return null, wrotePosition: " + currentPos + " fileSize: "
 18:         + this.fileSize);
 19:     return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
 20: }複製代碼

說明：插入消息到 MappedFile，並返回插入結果。
第 8 行：獲取須要寫入的字節緩衝區。爲何會有 writeBuffer != null 的判斷後，使用不一樣的字節緩衝區，見：FlushCommitLogService。
第 9 至 11 行：設置寫入 position，執行寫入，更新 wrotePosition(當前寫入位置，下次開始寫入開始位置)。

DefaultAppendMessageCallback#doAppend(...)

1: class DefaultAppendMessageCallback implements AppendMessageCallback {
  2:     // File at the end of the minimum fixed length empty
  3:     private static final int END_FILE_MIN_BLANK_LENGTH = 4 + 4;
  4:     /** 5: * 存儲在內存中的消息編號字節Buffer 6: */
  7:     private final ByteBuffer msgIdMemory;
  8:     /** 9: * Store the message content 10: * 存儲在內存中的消息字節Buffer 11: * 當消息傳遞到{@link #doAppend(long, ByteBuffer, int, MessageExtBrokerInner)}方法時，最終寫到該參數 12: */
 13:     private final ByteBuffer msgStoreItemMemory;
 14:     /** 15: * The maximum length of the message 16: * 消息最大長度 17: */
 18:     private final int maxMessageSize;
 19:     /** 20: * Build Message Key 21: * {@link #topicQueueTable}的key 22: * 計算方式：topic + "-" + queueId 23: */
 24:     private final StringBuilder keyBuilder = new StringBuilder();
 25:     /** 26: * host字節buffer 27: * 用於重複計算host的字節內容 28: */
 29:     private final ByteBuffer hostHolder = ByteBuffer.allocate(8);
 30: 
 31:     DefaultAppendMessageCallback(final int size) {
 32:         this.msgIdMemory = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
 33:         this.msgStoreItemMemory = ByteBuffer.allocate(size + END_FILE_MIN_BLANK_LENGTH);
 34:         this.maxMessageSize = size;
 35:     }
 36: 
 37:     public ByteBuffer getMsgStoreItemMemory() {
 38:         return msgStoreItemMemory;
 39:     }
 40: 
 41:     public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank, final MessageExtBrokerInner msgInner) {
 42:         // STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>
 43: 
 44:         // PHY OFFSET
 45:         long wroteOffset = fileFromOffset + byteBuffer.position();
 46: 
 47:         // 計算commitLog裏的msgId
 48:         this.resetByteBuffer(hostHolder, 8);
 49:         String msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(hostHolder), wroteOffset);
 50: 
 51:         // Record ConsumeQueue information 獲取隊列offset
 52:         keyBuilder.setLength(0);
 53:         keyBuilder.append(msgInner.getTopic());
 54:         keyBuilder.append('-');
 55:         keyBuilder.append(msgInner.getQueueId());
 56:         String key = keyBuilder.toString();
 57:         Long queueOffset = CommitLog.this.topicQueueTable.get(key);
 58:         if (null == queueOffset) {
 59:             queueOffset = 0L;
 60:             CommitLog.this.topicQueueTable.put(key, queueOffset);
 61:         }
 62: 
 63:         // Transaction messages that require special handling // TODO 疑問：用途
 64:         final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
 65:         switch (tranType) {
 66:             // Prepared and Rollback message is not consumed, will not enter the
 67:             // consumer queue
 68:             case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
 69:             case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
 70:                 queueOffset = 0L;
 71:                 break;
 72:             case MessageSysFlag.TRANSACTION_NOT_TYPE:
 73:             case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
 74:             default:
 75:                 break;
 76:         }
 77: 
 78:         // 計算消息長度
 79:         final byte[] propertiesData =
 80:             msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
 81:         final int propertiesLength = propertiesData == null ? 0 : propertiesData.length;
 82:         if (propertiesLength > Short.MAX_VALUE) {
 83:             log.warn("putMessage message properties length too long. length={}", propertiesData.length);
 84:             return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
 85:         }
 86:         final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
 87:         final int topicLength = topicData.length;
 88:         final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
 89:         final int msgLen = calMsgLength(bodyLength, topicLength, propertiesLength);
 90:         // Exceeds the maximum message
 91:         if (msgLen > this.maxMessageSize) {
 92:             CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLength
 93:                 + ", maxMessageSize: " + this.maxMessageSize);
 94:             return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
 95:         }
 96: 
 97:         // Determines whether there is sufficient(足夠) free space
 98:         if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
 99:             this.resetByteBuffer(this.msgStoreItemMemory, maxBlank);
100:             // 1 TOTAL_SIZE
101:             this.msgStoreItemMemory.putInt(maxBlank);
102:             // 2 MAGIC_CODE
103:             this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
104:             // 3 The remaining space may be any value
105:             //
106: 
107:             // Here the length of the specially set maxBlank
108:             final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
109:             byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank);
110:             return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(),
111:                 queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
112:         }
113: 
114:         // Initialization of storage space
115:         this.resetByteBuffer(msgStoreItemMemory, msgLen);
116:         // 1 TOTAL_SIZE
117:         this.msgStoreItemMemory.putInt(msgLen);
118:         // 2 MAGIC_CODE
119:         this.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);
120:         // 3 BODY_CRC
121:         this.msgStoreItemMemory.putInt(msgInner.getBodyCRC());
122:         // 4 QUEUE_ID
123:         this.msgStoreItemMemory.putInt(msgInner.getQueueId());
124:         // 5 FLAG
125:         this.msgStoreItemMemory.putInt(msgInner.getFlag());
126:         // 6 QUEUE_OFFSET
127:         this.msgStoreItemMemory.putLong(queueOffset);
128:         // 7 PHYSICAL_OFFSET
129:         this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());
130:         // 8 SYS_FLAG
131:         this.msgStoreItemMemory.putInt(msgInner.getSysFlag());
132:         // 9 BORN_TIMESTAMP
133:         this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
134:         // 10 BORN_HOST
135:         this.resetByteBuffer(hostHolder, 8);
136:         this.msgStoreItemMemory.put(msgInner.getBornHostBytes(hostHolder));
137:         // 11 STORE_TIMESTAMP
138:         this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
139:         // 12 STORE_HOST_ADDRESS
140:         this.resetByteBuffer(hostHolder, 8);
141:         this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(hostHolder));
142:         //this.msgStoreItemMemory.put(msgInner.getStoreHostBytes());
143:         // 13 RECONSUME_TIMES
144:         this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
145:         // 14 Prepared Transaction Offset
146:         this.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());
147:         // 15 BODY
148:         this.msgStoreItemMemory.putInt(bodyLength);
149:         if (bodyLength > 0)
150:             this.msgStoreItemMemory.put(msgInner.getBody());
151:         // 16 TOPIC
152:         this.msgStoreItemMemory.put((byte) topicLength);
153:         this.msgStoreItemMemory.put(topicData);
154:         // 17 PROPERTIES
155:         this.msgStoreItemMemory.putShort((short) propertiesLength);
156:         if (propertiesLength > 0)
157:             this.msgStoreItemMemory.put(propertiesData);
158: 
159:         final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
160:         // Write messages to the queue buffer
161:         byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);
162: 
163:         AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId,
164:             msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
165: 
166:         switch (tranType) {
167:             case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
168:             case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
169:                 break;
170:             case MessageSysFlag.TRANSACTION_NOT_TYPE:
171:             case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
172:                 // The next update ConsumeQueue information 更新隊列的offset
173:                 CommitLog.this.topicQueueTable.put(key, ++queueOffset);
174:                 break;
175:             default:
176:                 break;
177:         }
178:         return result;
179:     }
180: 
181:     /** 182: * 重置字節緩衝區 183: * 184: * @param byteBuffer 字節緩衝區 185: * @param limit 長度 186: */
187:     private void resetByteBuffer(final ByteBuffer byteBuffer, final int limit) {
188:         byteBuffer.flip();
189:         byteBuffer.limit(limit);
190:     }
191: }複製代碼

說明：插入消息到字節緩衝區。
第 45 行：計算物理位置。在 CommitLog 的順序存儲位置。
第 47 至 49 行：計算 CommitLog 裏的 offsetMsgId。這裏必定要和 msgId 區分開。

		計算方式	長度
offsetMsgId	Broker存儲時生成	Hex(storeHostBytes, wroteOffset)	32
msgId	Client發送消息時生成	Hex(進程編號, IP, ClassLoader, startTime, currentTime, 自增序列)	32	《RocketMQ 源碼分析 —— Message 基礎》

第 51 至 61 行：獲取隊列位置(offset)。
第 78 至 95 行：計算消息總長度。
第 98 至 112 行：當文件剩餘空間不足時，寫入 BLANK 佔位，返回結果。
第 114 至 161 行：寫入 MESSAGE 。
第 173 行：更新隊列位置(offset)。

FlushCommitLogService

線程服務	場景	插入消息性能
CommitRealTimeService	異步刷盤 && 開啓內存字節緩衝區	第一
FlushRealTimeService	異步刷盤 && 關閉內存字節緩衝區	第二
GroupCommitService	同步刷盤	第三

MappedFile#落盤

方式
方式一	寫入內存字節緩衝區(writeBuffer)	從內存字節緩衝區(write buffer)提交(commit)到文件通道(fileChannel)	文件通道(fileChannel)flush
方式二		寫入映射文件字節緩衝區(mappedByteBuffer)	映射文件字節緩衝區(mappedByteBuffer)flush

flush相關代碼

考慮到寫入性能，知足 flushLeastPages * OS_PAGE_SIZE 才進行 flush。

1: /** 2: * flush 3: * 4: * @param flushLeastPages flush最小頁數 5: * @return The current flushed position 6: */
  7: public int flush(final int flushLeastPages) {
  8:     if (this.isAbleToFlush(flushLeastPages)) {
  9:         if (this.hold()) {
 10:             int value = getReadPosition();
 11: 
 12:             try {
 13:                 //We only append data to fileChannel or mappedByteBuffer, never both.
 14:                 if (writeBuffer != null || this.fileChannel.position() != 0) {
 15:                     this.fileChannel.force(false);
 16:                 } else {
 17:                     this.mappedByteBuffer.force();
 18:                 }
 19:             } catch (Throwable e) {
 20:                 log.error("Error occurred when force data to disk.", e);
 21:             }
 22: 
 23:             this.flushedPosition.set(value);
 24:             this.release();
 25:         } else {
 26:             log.warn("in flush, hold failed, flush offset = " + this.flushedPosition.get());
 27:             this.flushedPosition.set(getReadPosition());
 28:         }
 29:     }
 30:     return this.getFlushedPosition();
 31: }
 32: 
 33: /** 34: * 是否可以flush。知足以下條件任意條件： 35: * 1. 映射文件已經寫滿 36: * 2. flushLeastPages > 0 && 未flush部分超過flushLeastPages 37: * 3. flushLeastPages = 0 && 有新寫入部分 38: * 39: * @param flushLeastPages flush最小分頁 40: * @return 是否可以寫入 41: */
 42: private boolean isAbleToFlush(final int flushLeastPages) {
 43:     int flush = this.flushedPosition.get();
 44:     int write = getReadPosition();
 45: 
 46:     if (this.isFull()) {
 47:         return true;
 48:     }
 49: 
 50:     if (flushLeastPages > 0) {
 51:         return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= flushLeastPages;
 52:     }
 53: 
 54:     return write > flush;
 55: }複製代碼

commit相關代碼：

考慮到寫入性能，知足 commitLeastPages * OS_PAGE_SIZE 才進行 commit。

1: /** 2: * commit 3: * 當{@link #writeBuffer}爲null時，直接返回{@link #wrotePosition} 4: * 5: * @param commitLeastPages commit最小頁數 6: * @return 當前commit位置 7: */
  8: public int commit(final int commitLeastPages) {
  9:     if (writeBuffer == null) {
 10:         //no need to commit data to file channel, so just regard wrotePosition as committedPosition.
 11:         return this.wrotePosition.get();
 12:     }
 13:     if (this.isAbleToCommit(commitLeastPages)) {
 14:         if (this.hold()) {
 15:             commit0(commitLeastPages);
 16:             this.release();
 17:         } else {
 18:             log.warn("in commit, hold failed, commit offset = " + this.committedPosition.get());
 19:         }
 20:     }
 21: 
 22:     // All dirty data has been committed to FileChannel. 寫到文件尾時，回收writeBuffer。
 23:     if (writeBuffer != null && this.transientStorePool != null && this.fileSize == this.committedPosition.get()) {
 24:         this.transientStorePool.returnBuffer(writeBuffer);
 25:         this.writeBuffer = null;
 26:     }
 27: 
 28:     return this.committedPosition.get();
 29: }
 30: 
 31: /** 32: * commit實現，將writeBuffer寫入fileChannel。 33: * @param commitLeastPages commit最小頁數。用不上該參數 34: */
 35: protected void commit0(final int commitLeastPages) {
 36:     int writePos = this.wrotePosition.get();
 37:     int lastCommittedPosition = this.committedPosition.get();
 38: 
 39:     if (writePos - this.committedPosition.get() > 0) {
 40:         try {
 41:             // 設置須要寫入的byteBuffer
 42:             ByteBuffer byteBuffer = writeBuffer.slice();
 43:             byteBuffer.position(lastCommittedPosition);
 44:             byteBuffer.limit(writePos);
 45:             // 寫入fileChannel
 46:             this.fileChannel.position(lastCommittedPosition);
 47:             this.fileChannel.write(byteBuffer);
 48:             // 設置position
 49:             this.committedPosition.set(writePos);
 50:         } catch (Throwable e) {
 51:             log.error("Error occurred when commit data to FileChannel.", e);
 52:         }
 53:     }
 54: }
 55: 
 56: /** 57: * 是否可以commit。知足以下條件任意條件： 58: * 1. 映射文件已經寫滿 59: * 2. commitLeastPages > 0 && 未commit部分超過commitLeastPages 60: * 3. commitLeastPages = 0 && 有新寫入部分 61: * 62: * @param commitLeastPages commit最小分頁 63: * @return 是否可以寫入 64: */
 65: protected boolean isAbleToCommit(final int commitLeastPages) {
 66:     int flush = this.committedPosition.get();
 67:     int write = this.wrotePosition.get();
 68: 
 69:     if (this.isFull()) {
 70:         return true;
 71:     }
 72: 
 73:     if (commitLeastPages > 0) {
 74:         return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= commitLeastPages;
 75:     }
 76: 
 77:     return write > flush;
 78: }複製代碼

FlushRealTimeService

消息插入成功時，異步刷盤時使用。

1: class FlushRealTimeService extends FlushCommitLogService {
  2:     /** 3: * 最後flush時間戳 4: */
  5:     private long lastFlushTimestamp = 0;
  6:     /** 7: * print計時器。 8: * 知足print次數時，調用{@link #printFlushProgress()} 9: */
 10:     private long printTimes = 0;
 11: 
 12:     public void run() {
 13:         CommitLog.log.info(this.getServiceName() + " service started");
 14: 
 15:         while (!this.isStopped()) {
 16:             boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed();
 17:             int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
 18:             int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();
 19:             int flushPhysicQueueThoroughInterval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval();
 20: 
 21:             // Print flush progress
 22:             // 當時間知足flushPhysicQueueThoroughInterval時，即便寫入的數量不足flushPhysicQueueLeastPages，也進行flush
 23:             boolean printFlushProgress = false;
 24:             long currentTimeMillis = System.currentTimeMillis();
 25:             if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {
 26:                 this.lastFlushTimestamp = currentTimeMillis;
 27:                 flushPhysicQueueLeastPages = 0;
 28:                 printFlushProgress = (printTimes++ % 10) == 0;
 29:             }
 30: 
 31:             try {
 32:                 // 等待執行
 33:                 if (flushCommitLogTimed) {
 34:                     Thread.sleep(interval);
 35:                 } else {
 36:                     this.waitForRunning(interval);
 37:                 }
 38: 
 39:                 if (printFlushProgress) {
 40:                     this.printFlushProgress();
 41:                 }
 42: 
 43:                 // flush commitLog
 44:                 long begin = System.currentTimeMillis();
 45:                 CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages);
 46:                 long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
 47:                 if (storeTimestamp > 0) {
 48:                     CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
 49:                 }
 50:                 long past = System.currentTimeMillis() - begin;
 51:                 if (past > 500) {
 52:                     log.info("Flush data to disk costs {} ms", past);
 53:                 }
 54:             } catch (Throwable e) {
 55:                 CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
 56:                 this.printFlushProgress();
 57:             }
 58:         }
 59: 
 60:         // Normal shutdown, to ensure that all the flush before exit
 61:         boolean result = false;
 62:         for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
 63:             result = CommitLog.this.mappedFileQueue.flush(0);
 64:             CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
 65:         }
 66: 
 67:         this.printFlushProgress();
 68: 
 69:         CommitLog.log.info(this.getServiceName() + " service end");
 70:     }
 71: 
 72:     @Override
 73:     public String getServiceName() {
 74:         return FlushRealTimeService.class.getSimpleName();
 75:     }
 76: 
 77:     private void printFlushProgress() {
 78:         // CommitLog.log.info("how much disk fall behind memory, "
 79:         // + CommitLog.this.mappedFileQueue.howMuchFallBehind());
 80:     }
 81: 
 82:     @Override
 83:     @SuppressWarnings("SpellCheckingInspection")
 84:     public long getJointime() {
 85:         return 1000 * 60 * 5;
 86:     }
 87: }複製代碼

說明：實時 flush線程服務，調用 MappedFile#flush 相關邏輯。
第 23 至 29 行：每 flushPhysicQueueThoroughInterval 週期，執行一次 flush 。由於不是每次循環到都能知足 flushCommitLogLeastPages 大小，所以，須要必定週期進行一次強制 flush 。固然，不能每次循環都去執行強制 flush，這樣性能較差。
第 33 行至 37 行：根據 flushCommitLogTimed 參數，能夠選擇每次循環是固定週期仍是等待喚醒。默認配置是後者，因此，每次插入消息完成，會去調用 commitLogService.wakeup() 。
第 45 行：調用 MappedFile 進行 flush。
第 61 至 65 行：Broker 關閉時，強制 flush，避免有未刷盤的數據。

CommitRealTimeService

消息插入成功時，異步刷盤時使用。
和 FlushRealTimeService 相似，性能更好。

1: class CommitRealTimeService extends FlushCommitLogService {
  2: 
  3:     /** 4: * 最後 commit 時間戳 5: */
  6:     private long lastCommitTimestamp = 0;
  7: 
  8:     @Override
  9:     public String getServiceName() {
 10:         return CommitRealTimeService.class.getSimpleName();
 11:     }
 12: 
 13:     @Override
 14:     public void run() {
 15:         CommitLog.log.info(this.getServiceName() + " service started");
 16:         while (!this.isStopped()) {
 17:             int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog();
 18:             int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();
 19:             int commitDataThoroughInterval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval();
 20: 
 21:             // 當時間知足commitDataThoroughInterval時，即便寫入的數量不足commitDataLeastPages，也進行flush
 22:             long begin = System.currentTimeMillis();
 23:             if (begin >= (this.lastCommitTimestamp + commitDataThoroughInterval)) {
 24:                 this.lastCommitTimestamp = begin;
 25:                 commitDataLeastPages = 0;
 26:             }
 27: 
 28:             try {
 29:                 // commit
 30:                 boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);
 31:                 long end = System.currentTimeMillis();
 32:                 if (!result) { // TODO 疑問：未寫入成功，爲啥要喚醒flushCommitLogService
 33:                     this.lastCommitTimestamp = end; // result = false means some data committed.
 34:                     //now wake up flush thread.
 35:                     flushCommitLogService.wakeup();
 36:                 }
 37: 
 38:                 if (end - begin > 500) {
 39:                     log.info("Commit data to file costs {} ms", end - begin);
 40:                 }
 41: 
 42:                 // 等待執行
 43:                 this.waitForRunning(interval);
 44:             } catch (Throwable e) {
 45:                 CommitLog.log.error(this.getServiceName() + " service has exception. ", e);
 46:             }
 47:         }
 48: 
 49:         boolean result = false;
 50:         for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
 51:             result = CommitLog.this.mappedFileQueue.commit(0);
 52:             CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
 53:         }
 54:         CommitLog.log.info(this.getServiceName() + " service end");
 55:     }
 56: }複製代碼

GroupCommitService

消息插入成功時，同步刷盤時使用。

1: class GroupCommitService extends FlushCommitLogService {
  2:     /** 3: * 寫入請求隊列 4: */
  5:     private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<>();
  6:     /** 7: * 讀取請求隊列 8: */
  9:     private volatile List<GroupCommitRequest> requestsRead = new ArrayList<>();
 10: 
 11:     /** 12: * 添加寫入請求 13: * 14: * @param request 寫入請求 15: */
 16:     public synchronized void putRequest(final GroupCommitRequest request) {
 17:         // 添加寫入請求
 18:         synchronized (this.requestsWrite) {
 19:             this.requestsWrite.add(request);
 20:         }
 21:         // 切換讀寫隊列
 22:         if (hasNotified.compareAndSet(false, true)) {
 23:             waitPoint.countDown(); // notify
 24:         }
 25:     }
 26: 
 27:     /** 28: * 切換讀寫隊列 29: */
 30:     private void swapRequests() {
 31:         List<GroupCommitRequest> tmp = this.requestsWrite;
 32:         this.requestsWrite = this.requestsRead;
 33:         this.requestsRead = tmp;
 34:     }
 35: 
 36:     private void doCommit() {
 37:         synchronized (this.requestsRead) {
 38:             if (!this.requestsRead.isEmpty()) {
 39:                 for (GroupCommitRequest req : this.requestsRead) {
 40:                     // There may be a message in the next file, so a maximum of
 41:                     // two times the flush (可能批量提交的messages，分佈在兩個MappedFile)
 42:                     boolean flushOK = false;
 43:                     for (int i = 0; i < 2 && !flushOK; i++) {
 44:                         // 是否知足須要flush條件，即請求的offset超過flush的offset
 45:                         flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
 46:                         if (!flushOK) {
 47:                             CommitLog.this.mappedFileQueue.flush(0);
 48:                         }
 49:                     }
 50:                     // 喚醒等待請求
 51:                     req.wakeupCustomer(flushOK);
 52:                 }
 53: 
 54:                 long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
 55:                 if (storeTimestamp > 0) {
 56:                     CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
 57:                 }
 58: 
 59:                 // 清理讀取隊列
 60:                 this.requestsRead.clear();
 61:             } else {
 62:                 // Because of individual messages is set to not sync flush, it
 63:                 // will come to this process 不合法的請求，好比message上未設置isWaitStoreMsgOK。
 64:                 // 走到此處的邏輯，至關於發送一條消息，落盤一條消息，實際無批量提交的效果。
 65:                 CommitLog.this.mappedFileQueue.flush(0);
 66:             }
 67:         }
 68:     }
 69: 
 70:     public void run() {
 71:         CommitLog.log.info(this.getServiceName() + " service started");
 72: 
 73:         while (!this.isStopped()) {
 74:             try {
 75:                 this.waitForRunning(10);
 76:                 this.doCommit();
 77:             } catch (Exception e) {
 78:                 CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
 79:             }
 80:         }
 81: 
 82:         // Under normal circumstances shutdown, wait for the arrival of the
 83:         // request, and then flush
 84:         try {
 85:             Thread.sleep(10);
 86:         } catch (InterruptedException e) {
 87:             CommitLog.log.warn("GroupCommitService Exception, ", e);
 88:         }
 89: 
 90:         synchronized (this) {
 91:             this.swapRequests();
 92:         }
 93: 
 94:         this.doCommit();
 95: 
 96:         CommitLog.log.info(this.getServiceName() + " service end");
 97:     }
 98: 
 99:     /** 100: * 每次執行完，切換讀寫隊列 101: */
102:     @Override
103:     protected void onWaitEnd() {
104:         this.swapRequests();
105:     }
106: 
107:     @Override
108:     public String getServiceName() {
109:         return GroupCommitService.class.getSimpleName();
110:     }
111: 
112:     @Override
113:     public long getJointime() {
114:         return 1000 * 60 * 5;
115:     }
116: }複製代碼

說明：批量寫入線程服務。
第 16 至 25 行：添加寫入請求。方法設置了 sync 的緣由：this.requestsWrite 會和 this.requestsRead 不斷交換，沒法保證穩定的同步。
第 27 至 34 行：讀寫隊列交換。
第 38 至 60 行：循環寫入隊列，進行 flush。
- 第 43 行：考慮到有可能每次循環的消息寫入的消息，可能分佈在兩個 MappedFile(寫第N個消息時，MappedFile 已滿，建立了一個新的)，因此須要有循環2次。
- 第 51 行：喚醒等待寫入請求線程，經過 CountDownLatch 實現。
第 61 至 66 行：直接刷盤。此處是因爲發送的消息的 isWaitStoreMsgOK 未設置成 TRUE ，致使未走批量提交。
第 73 至 80 行：每 10ms 執行一次批量提交。固然，若是 wakeup() 時，則會當即進行一次批量提交。當 Broker 設置成同步落盤 && 消息 isWaitStoreMsgOK=true，消息須要略大於 10ms 才能發送成功。固然，性能相對異步落盤較差，可靠性更高，須要咱們在實際使用時去取捨。