一、簡介
RocketMQ 是阿里巴巴開源的分布式消息中間件，它借鑒了 Kafka 實現，支持消息訂閱與發布、順序消息、事務消息、定時消息、消息回溯、死信隊列等功能。RocketMQ 架構上主要分為四部分，如下圖所示：
-
Producer：消息生產者，支持分布式集群方式部署。
-
Consumer：消息消費者，支持分布式集群方式部署。
-
NameServer：名字服務，是一個非常簡單的 Topic 路由注冊中心，支持 Broker 的動態注冊與發現，Producer 和 Consumer 通過 NameServer 動態感知 Broker 的路由信息。
-
Broker：Broker 主要負責消息的存儲、轉發和查詢。
本文基於 Apache RocketMQ 4.9.1 版本剖析 Broker 中的消息存儲模塊是如何設計的。
二、存儲架構
RocketMQ 的消息文件路徑如圖所示。
CommitLog
消息主體以及元數據的存儲主體，存儲 Producer 端寫入的消息主體內容，消息內容不是定長的。單個文件大小默認1G， 文件名長度為 20 位，左邊補零，剩余為起始偏移量，比如 00000000000000000000 代表了第一個文件，起始偏移量為 0，文件大小為 1G=1073741824；當第一個文件寫滿了，第二個文件為 00000000001073741824，起始偏移量為 1073741824，以此類推。
ConsumeQueue
消息消費隊列，Consumequeue 文件可以看成是基於 CommitLog 的索引文件。ConsumeQueue 文件采取定長設計，每一個條目共 20 個字節，分別為 8 字節的 CommitLog 物理偏移量、4 字節的消息長度、8 字節 tag hashcode，單個文件由 30W 個條目組成，可以像數組一樣隨機訪問每一個條目，每個 ConsumeQueue 文件大小約 5.72M。
IndexFile
索引文件，提供了一種可以通過 key 或時間區間來查詢消息的方法。單個 IndexFile 文件大小約為 400M，一個 IndexFile 可以保存 2000W 個索引，IndexFile 的底層存儲設計類似 JDK 的 HashMap 數據結構。
其他文件：包括 config 文件夾，存放運行時配置信息；abort 文件，說明 Broker 是否正常關閉；checkpoint 文件，存儲 Commitlog、ConsumeQueue、Index 文件最后一次刷盤時間戳。這些不在本文討論的范圍。
同 Kafka 相比，Kafka 每個 Topic 的每個 partition 對應一個文件，順序寫入，定時刷盤。但一旦單個 Broker 的 Topic 過多，順序寫將退化為隨機寫。而 RocketMQ 單個 Broker 所有 Topic 在同一個 CommitLog 中順序寫，是能夠保證嚴格順序寫。RocketMQ 讀取消息需要從 ConsumeQueue 中拿到消息實際物理偏移再去 CommitLog 讀取消息內容，會造成隨機讀取。
2.1 Page Cache 和 mmap
在正式介紹 Broker 消息存儲模塊實現前，先說明下 Page Cache 和 mmap 這兩個概念。
Page Cache 是 OS 對文件的緩存，用於加速對文件的讀寫。一般來說，程序對文件進行順序讀寫的速度幾乎接近於內存的讀寫速度，主要原因就是由於 OS 使用 Page Cache 機制對讀寫訪問操作進行了性能優化，將一部分的內存用作 Page Cache。對於數據的寫入，OS 會先寫入至 Cache 內，隨后通過異步的方式由 pdflush 內核線程將 Cache 內的數據刷盤至物理磁盤上。對於數據的讀取，如果一次讀取文件時出現未命中 Page Cache 的情況，OS 從物理磁盤上訪問讀取文件的同時，會順序對其他相鄰塊的數據文件進行預讀取。
mmap 是將磁盤上的物理文件直接映射到用戶態的內存地址中，減少了傳統 IO 將磁盤文件數據在操作系統內核地址空間的緩沖區和用戶應用程序地址空間的緩沖區之間來回進行拷貝的性能開銷。Java NIO 中的 FileChannel 提供了 map() 方法可以實現 mmap。FileChannel (文件通道)和 mmap (內存映射) 讀寫性能比較可以參照這篇文章。
2.2 Broker 模塊
下圖是 Broker 存儲架構圖，展示了 Broker 模塊從收到消息到返回響應業務流轉過程。
業務接入層：RocketMQ 基於 Netty 的 Reactor 多線程模型實現了底層通信。Reactor 主線程池 eventLoopGroupBoss 負責創建 TCP 連接，默認只有一個線程。連接建立后，再丟給 Reactor 子線程池 eventLoopGroupSelector 進行讀寫事件的處理。
defaultEventExecutorGroup 負責 SSL 驗證、編解碼、空閑檢查、網絡連接管理。然后根據 RomotingCommand 的業務請求碼 code 去 processorTable 這個本地緩存變量中找到對應的 processor，封裝成 task 任務后，提交給對應的業務 processor 處理線程池來執行。Broker 模塊通過這四級線程池提升系統吞吐量。
業務處理層：處理各種通過 RPC 調用過來的業務請求，其中：
-
SendMessageProcessor 負責處理 Producer 發送消息的請求；
-
PullMessageProcessor 負責處理 Consumer 消費消息的請求；
-
QueryMessageProcessor 負責處理按照消息 Key 等查詢消息的請求。
存儲邏輯層：DefaultMessageStore 是 RocketMQ 的存儲邏輯核心類，提供消息存儲、讀取、刪除等能力。
文件映射層：把 Commitlog、ConsumeQueue、IndexFile 文件映射為存儲對象 MappedFile。
數據傳輸層：支持基於 mmap 內存映射進行讀寫消息，同時也支持基於 mmap 進行讀取消息、堆外內存寫入消息的方式進行讀寫消息。
下面章節將從源碼角度來剖析 RocketMQ 是如何實現高性能存儲。
三、消息寫入
以單個消息生產為例，消息寫入時序邏輯如下圖，業務邏輯如上文 Broker 存儲架構所示在各層之間進行流轉。
最底層消息寫入核心代碼在 CommitLog 的 asyncPutMessage 方法中，主要分為獲取 MappedFile、往緩沖區寫消息、提交刷盤請求三步。需要注意的是在這三步前后有自旋鎖或 ReentrantLock 的加鎖、釋放鎖，保證單個 Broker 寫消息是串行的。
//org.apache.rocketmq.store.CommitLog::asyncPutMessage
public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
...
putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
try {
//獲取最新的 MappedFile
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
...
//向緩沖區寫消息
result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
...
//提交刷盤請求
CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);
...
} finally {
putMessageLock.unlock();
}
...
}
下面介紹這三步具體做了什么事情。
3.1 MappedFile 初始化
在 Broker 初始化時會啟動管理 MappedFile 創建的 AllocateMappedFileService 異步線程。消息處理線程 和 AllocateMappedFileService 線程通過隊列 requestQueue 關聯。
消息寫入時調用 AllocateMappedFileService 的 putRequestAndReturnMappedFile 方法往 requestQueue 放入提交創建 MappedFile 請求，這邊會同時構建兩個 AllocateRequest 放入隊列。
AllocateMappedFileService 線程循環從 requestQueue 獲取 AllocateRequest 來創建 MappedFile。消息處理線程通過 CountDownLatch 等待獲取第一個 MappedFile 創建成功就返回。
當消息處理線程需要再次創建 MappedFile 時，此時可以直接獲取之前已預創建的 MappedFile。這樣通過預創建 MappedFile ，減少文件創建等待時間。
//org.apache.rocketmq.store.AllocateMappedFileService::putRequestAndReturnMappedFile
public MappedFile putRequestAndReturnMappedFile(String nextFilePath, String nextNextFilePath, int fileSize) {
//請求創建 MappedFile
AllocateRequest nextReq = new AllocateRequest(nextFilePath, fileSize);
boolean nextPutOK = this.requestTable.putIfAbsent(nextFilePath, nextReq) == null;
...
//請求預先創建下一個 MappedFile
AllocateRequest nextNextReq = new AllocateRequest(nextNextFilePath, fileSize);
boolean nextNextPutOK = this.requestTable.putIfAbsent(nextNextFilePath, nextNextReq) == null;
...
//獲取本次創建 MappedFile
AllocateRequest result = this.requestTable.get(nextFilePath);
...
}
//org.apache.rocketmq.store.AllocateMappedFileService::run
public void run() {
..
while (!this.isStopped() && this.mmapOperation()) {
}
...
}
//org.apache.rocketmq.store.AllocateMappedFileService::mmapOperation
private boolean mmapOperation() {
...
//從隊列獲取 AllocateRequest
req = this.requestQueue.take();
...
//判斷是否開啟堆外內存池
if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
//開啟堆外內存的 MappedFile
mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
} else {
//普通 MappedFile
mappedFile = new MappedFile(req.getFilePath(), req.getFileSize());
}
...
//MappedFile 預熱
if (mappedFile.getFileSize() >= this.messageStore.getMessageStoreConfig()
.getMappedFileSizeCommitLog()
&&
this.messageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
mappedFile.warmMappedFile(this.messageStore.getMessageStoreConfig().getFlushDiskType(),
this.messageStore.getMessageStoreConfig().getFlushLeastPagesWhenWarmMapedFile());
}
req.setMappedFile(mappedFile);
...
}
每次新建普通 MappedFile 請求，都會創建 mappedByteBuffer，下面代碼展示了 Java mmap 是如何實現的。
//org.apache.rocketmq.store.MappedFile::init
private void init(final String fileName, final int fileSize) throws IOException {
...
this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
...
}
如果開啟堆外內存，即 transientStorePoolEnable = true 時，mappedByteBuffer 只是用來讀消息，堆外內存用來寫消息，從而實現對於消息的讀寫分離。堆外內存對象不是每次新建 MappedFile 都需要創建，而是系統啟動時根據堆外內存池大小就初始化好了。每個堆外內存 DirectByteBuffer 都與 CommitLog 文件大小相同，通過鎖定住該堆外內存，確保不會被置換到虛擬內存中去。
//org.apache.rocketmq.store.TransientStorePool
public void init() {
for (int i = 0; i < poolSize; i++) {
//分配與 CommitLog 文件大小相同的堆外內存
ByteBuffer byteBuffer = ByteBuffer.allocateDirect(fileSize);
final long address = ((DirectBuffer) byteBuffer).address();
Pointer pointer = new Pointer(address);
//鎖定堆外內存，確保不會被置換到虛擬內存中去
LibC.INSTANCE.mlock(pointer, new NativeLong(fileSize));
availableBuffers.offer(byteBuffer);
}
}
上面的 mmapOperation 方法中有段 MappedFile 預熱邏輯。為什么需要文件預熱呢？文件預熱怎么做的呢？
因為通過 mmap 映射，只是建立了進程虛擬內存地址與物理內存地址之間的映射關系，並沒有將 Page Cache 加載至內存。讀寫數據時如果沒有命中寫 Page Cache 則發生缺頁中斷，從磁盤重新加載數據至內存，這樣會影響讀寫性能。為了防止缺頁異常，阻止操作系統將相關的內存頁調度到交換空間（swap space），RocketMQ 通過對文件預熱，文件預熱實現如下。
//org.apache.rocketmq.store.MappedFile::warmMappedFile
public void warmMappedFile(FlushDiskType type, int pages) {
ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
int flush = 0;
//通過寫入 1G 的字節 0 來讓操作系統分配物理內存空間，如果沒有填充值，操作系統不會實際分配物理內存，防止在寫入消息時發生缺頁異常
for (int i = 0, j = 0; i < this.fileSize; i += MappedFile.OS_PAGE_SIZE, j++) {
byteBuffer.put(i, (byte) 0);
// force flush when flush disk type is sync
if (type == FlushDiskType.SYNC_FLUSH) {
if ((i / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE) >= pages) {
flush = i;
mappedByteBuffer.force();
}
}
//prevent gc
if (j % 1000 == 0) {
Thread.sleep(0);
}
}
//force flush when prepare load finished
if (type == FlushDiskType.SYNC_FLUSH) {
mappedByteBuffer.force();
}
...
this.mlock();
}
//org.apache.rocketmq.store.MappedFile::mlock
public void mlock() {
final long beginTime = System.currentTimeMillis();
final long address = ((DirectBuffer) (this.mappedByteBuffer)).address();
Pointer pointer = new Pointer(address);
//通過系統調用 mlock 鎖定該文件的 Page Cache，防止其被交換到 swap 空間
int ret = LibC.INSTANCE.mlock(pointer, new NativeLong(this.fileSize));
//通過系統調用 madvise 給操作系統建議，說明該文件在不久的將來要被訪問
int ret = LibC.INSTANCE.madvise(pointer, new NativeLong(this.fileSize), LibC.MADV_WILLNEED);
}
綜上所述，RocketMQ 每次都預創建一個文件來減少文件創建延遲，通過文件預熱避免了讀寫時缺頁異常。
3.2 消息寫入
3.2.1 寫入 CommitLog
CommitLog 中每條消息存儲的邏輯視圖如下圖所示， TOTALSIZE 是整個消息占用存儲空間大小。
下面表格說明下每條消息包含哪些字段，以及這些字段占用空間大小和字段簡介。
消息的寫入是調用MappedFile 的 appendMessagesInner方法。
//org.apache.rocketmq.store.MappedFile::appendMessagesInner
public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb,
PutMessageContext putMessageContext) {
//判斷使用 DirectBuffer 還是 MappedByteBuffer 進行寫操作
ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
..
byteBuffer.position(currentPos);
AppendMessageResult result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos,
(MessageExtBrokerInner) messageExt, putMessageContext);
..
return result;
}
//org.apache.rocketmq.store.CommitLog::doAppend
public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
final MessageExtBrokerInner msgInner, PutMessageContext putMessageContext) {
...
ByteBuffer preEncodeBuffer = msgInner.getEncodedBuff();
...
//這邊只是將消息寫入緩沖區，還未實際刷盤
byteBuffer.put(preEncodeBuffer);
msgInner.setEncodedBuff(null);
...
return result;
}
至此，消息最終寫入 ByteBuffer，還沒有持久到磁盤，具體何時持久化，下一小節會具體講刷盤機制。這邊有個疑問 ConsumeQueue 和 IndexFile 是怎么寫入的？
答案是在存儲架構圖中存儲邏輯層的 ReputMessageService。MessageStore 在初始化的時候，會啟動一個 ReputMessageService 異步線程，它啟動后便會在循環中不斷調用 doReput 方法，用來通知 ConsumeQueue 和 IndexFile 進行更新。ConsumeQueue 和 IndexFile 之所以可以異步更新是因為 CommitLog 中保存了恢復 ConsumeQueue 和 IndexFile 所需隊列和 Topic 等信息，即使 Broker 服務異常宕機，Broker 重啟后可以根據 CommitLog 恢復 ConsumeQueue 和IndexFile。
//org.apache.rocketmq.store.DefaultMessageStore.ReputMessageService::run
public void run() {
...
while (!this.isStopped()) {
Thread.sleep(1);
this.doReput();
}
...
}
//org.apache.rocketmq.store.DefaultMessageStore.ReputMessageService::doReput
private void doReput() {
...
//獲取CommitLog中存儲的新消息
DispatchRequest dispatchRequest =
DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false);
int size = dispatchRequest.getBufferSize() == -1 ? dispatchRequest.getMsgSize() : dispatchRequest.getBufferSize();
if (dispatchRequest.isSuccess()) {
if (size > 0) {
//如果有新消息，則分別調用 CommitLogDispatcherBuildConsumeQueue、CommitLogDispatcherBuildIndex 進行構建 ConsumeQueue 和 IndexFile
DefaultMessageStore.this.doDispatch(dispatchRequest);
}
...
}
3.2.2 寫入 ConsumeQueue
如下圖所示，ConsumeQueue 每一條記錄共 20 個字節，分別為 8 字節的 CommitLog 物理偏移量、4 字節的消息長度、8字節 tag hashcode。
ConsumeQueue 記錄持久化邏輯如下。
//org.apache.rocketmq.store.ConsumeQueue::putMessagePositionInfo
private boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,
final long cqOffset) {
...
this.byteBufferIndex.flip();
this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE);
this.byteBufferIndex.putLong(offset);
this.byteBufferIndex.putInt(size);
this.byteBufferIndex.putLong(tagsCode);
final long expectLogicOffset = cqOffset * CQ_STORE_UNIT_SIZE;
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(expectLogicOffset);
if (mappedFile != null) {
...
return mappedFile.appendMessage(this.byteBufferIndex.array());
}
}
3.2.3 寫入 IndexFile
IndexFile 的文件邏輯結構如下圖所示，類似於 JDK 的 HashMap 的數組加鏈表結構。主要由 Header、Slot Table、Index Linked List 三部分組成。
Header：IndexFile 的頭部，占 40 個字節。主要包含以下字段：
-
beginTimestamp：該 IndexFile 文件中包含消息的最小存儲時間。
-
endTimestamp：該 IndexFile 文件中包含消息的最大存儲時間。
-
beginPhyoffset：該 IndexFile 文件中包含消息的最小 CommitLog 文件偏移量。
-
endPhyoffset：該 IndexFile 文件中包含消息的最大 CommitLog 文件偏移量。
-
hashSlotcount：該 IndexFile 文件中包含的 hashSlot 的總數。
-
indexCount：該 IndexFile 文件中已使用的 Index 條目個數。
Slot Table：默認包含 500w 個 Hash 槽，每個 Hash 槽存儲的是相同 hash 值的第一個 IndexItem 存儲位置 。
Index Linked List：默認最多包含 2000w 個 IndexItem。其組成如下所示：
-
Key Hash：消息 key 的 hash，當根據 key 搜索時比較的是其 hash，在之后會比較 key 本身。
-
CommitLog Offset：消息的物理位移。
-
Timestamp：該消息存儲時間與第一條消息的時間戳的差值。
-
Next Index Offset：發生 hash 沖突后保存的下一個 IndexItem 的位置。
Slot Table 中每個 hash 槽存放的是 IndexItem 在 Index Linked List 的位置，如果 hash 沖突時，新的 IndexItem 插入鏈表頭， 它的 Next Index Offset 中存放之前鏈表頭 IndexItem 位置，同時覆蓋 Slot Table 中的 hash 槽為最新 IndexItem 位置。代碼如下：
//org.apache.rocketmq.store.index.IndexFile::putKey
public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
int keyHash = indexKeyHashMethod(key);
int slotPos = keyHash % this.hashSlotNum;
int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
...
//從 Slot Table 獲取當前最新消息位置
int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
...
int absIndexPos =
IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
+ this.indexHeader.getIndexCount() * indexSize;
this.mappedByteBuffer.putInt(absIndexPos, keyHash);
this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
//存放之前鏈表頭 IndexItem 位置
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
//更新 Slot Table 中 hash 槽的值為最新消息位置
this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
if (this.indexHeader.getIndexCount() <= 1) {
this.indexHeader.setBeginPhyOffset(phyOffset);
this.indexHeader.setBeginTimestamp(storeTimestamp);
}
if (invalidIndex == slotValue) {
this.indexHeader.incHashSlotCount();
}
this.indexHeader.incIndexCount();
this.indexHeader.setEndPhyOffset(phyOffset);
this.indexHeader.setEndTimestamp(storeTimestamp);
return true;
...
}
綜上所述一個完整的消息寫入流程包括：同步寫入 Commitlog 文件緩存區，異步構建 ConsumeQueue、IndexFile 文件。
3.3 消息刷盤
RocketMQ 消息刷盤主要分為同步刷盤和異步刷盤。
(1) 同步刷盤：只有在消息真正持久化至磁盤后 RocketMQ 的 Broker 端才會真正返回給 Producer 端一個成功的 ACK 響應。同步刷盤對 MQ 消息可靠性來說是一種不錯的保障，但是性能上會有較大影響，一般金融業務使用該模式較多。
(2) 異步刷盤：能夠充分利用 OS 的 Page Cache 的優勢，只要消息寫入 Page Cache 即可將成功的 ACK 返回給 Producer 端。消息刷盤采用后台異步線程提交的方式進行，降低了讀寫延遲，提高了 MQ 的性能和吞吐量。異步刷盤包含開啟堆外內存和未開啟堆外內存兩種方式。
在 CommitLog 中提交刷盤請求時，會根據當前 Broker 相關配置決定是同步刷盤還是異步刷盤。
//org.apache.rocketmq.store.CommitLog::submitFlushRequest
public CompletableFuture<PutMessageStatus> submitFlushRequest(AppendMessageResult result, MessageExt messageExt) {
//同步刷盤
if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
if (messageExt.isWaitStoreMsgOK()) {
GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes(),
this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
service.putRequest(request);
return request.future();
} else {
service.wakeup();
return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
}
}
//異步刷盤
else {
if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
flushCommitLogService.wakeup();
} else {
//開啟堆外內存的異步刷盤
commitLogService.wakeup();
}
return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
}
}
GroupCommitService、FlushRealTimeService、CommitRealTimeService 三者繼承關系如圖；
GroupCommitService：同步刷盤線程。如下圖所示，消息寫入到 Page Cache 后通過 GroupCommitService 同步刷盤，消息處理線程阻塞等待刷盤結果。
//org.apache.rocketmq.store.CommitLog.GroupCommitService::run
public void run() {
...
while (!this.isStopped()) {
this.waitForRunning(10);
this.doCommit();
}
...
}
//org.apache.rocketmq.store.CommitLog.GroupCommitService::doCommit
private void doCommit() {
...
for (GroupCommitRequest req : this.requestsRead) {
boolean flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
for (int i = 0; i < 2 && !flushOK; i++) {
CommitLog.this.mappedFileQueue.flush(0);
flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
}
//喚醒等待刷盤完成的消息處理線程
req.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT);
}
...
}
//org.apache.rocketmq.store.MappedFile::flush
public int flush(final int flushLeastPages) {
if (this.isAbleToFlush(flushLeastPages)) {
...
//使用到了 writeBuffer 或者 fileChannel 的 position 不為 0 時用 fileChannel 進行強制刷盤
if (writeBuffer != null || this.fileChannel.position() != 0) {
this.fileChannel.force(false);
} else {
//使用 MappedByteBuffer 進行強制刷盤
this.mappedByteBuffer.force();
}
...
}
}
FlushRealTimeService：未開啟堆外內存的異步刷盤線程。如下圖所示，消息寫入到 Page Cache 后，消息處理線程立即返回，通過 FlushRealTimeService 異步刷盤。
//org.apache.rocketmq.store.CommitLog.FlushRealTimeService
public void run() {
...
//判斷是否需要周期性進行刷盤
if (flushCommitLogTimed) {
//固定休眠 interval 時間間隔
Thread.sleep(interval);
} else {
// 如果被喚醒就刷盤，非周期性刷盤
this.waitForRunning(interval);
}
...
// 這邊和 GroupCommitService 用的是同一個強制刷盤方法
CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages);
...
}
CommitRealTimeService：開啟堆外內存的異步刷盤線程。如下圖所示，消息處理線程把消息寫入到堆外內存后立即返回。后續先通過 CommitRealTimeService 把消息由堆外內存異步提交至 Page Cache，再由 FlushRealTimeService 線程異步刷盤。
注意：在消息異步提交至 Page Cache 后，業務就可以從 MappedByteBuffer 讀取到該消息。
消息寫入到堆外內存 writeBuffer 后，會通過 isAbleToCommit 方法判斷是否積累到至少提交頁數（默認4頁）。如果頁數達到最小提交頁數，則批量提交；否則還是駐留在堆外內存，這邊有丟失消息風險。通過這種批量操作，讀和寫的 Page Cahe 會間隔數頁，降低了 Page Cahe 讀寫沖突的概率，實現了讀寫分離。具體實現邏輯如下：
//org.apache.rocketmq.store.CommitLog.CommitRealTimeService
class CommitRealTimeService extends FlushCommitLogService {
@Override
public void run() {
while (!this.isStopped()) {
...
int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();
...
//把消息 commit 到內存緩沖區，最終調用的是 MappedFile::commit0 方法，只有達到最少提交頁數才能提交成功，否則還在堆外內存中
boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);
if (!result) {
//喚醒 flushCommitLogService，進行強制刷盤
flushCommitLogService.wakeup();
}
...
this.waitForRunning(interval);
}
}
}
//org.apache.rocketmq.store.MappedFile::commit0
protected void commit0() {
int writePos = this.wrotePosition.get();
int lastCommittedPosition = this.committedPosition.get();
//消息提交至 Page Cache，並未實際刷盤
if (writePos - lastCommittedPosition > 0) {
ByteBuffer byteBuffer = writeBuffer.slice();
byteBuffer.position(lastCommittedPosition);
byteBuffer.limit(writePos);
this.fileChannel.position(lastCommittedPosition);
this.fileChannel.write(byteBuffer);
this.committedPosition.set(writePos);
}
}
下面總結一下三種刷盤機制的使用場景及優缺點。
四、消息讀取
消息讀取邏輯相比寫入邏輯簡單很多，下面着重分析下根據 offset 查詢消息和根據 key 查詢消息是如何實現的。
4.1 根據 offset 查詢
讀取消息的過程就是先從 ConsumeQueue 中找到消息在 CommitLog 的物理偏移地址，然后再從 CommitLog 文件中讀取消息的實體內容。
//org.apache.rocketmq.store.DefaultMessageStore::getMessage
public GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset,
final int maxMsgNums,
final MessageFilter messageFilter) {
long nextBeginOffset = offset;
GetMessageResult getResult = new GetMessageResult();
final long maxOffsetPy = this.commitLog.getMaxOffset();
//找到對應的 ConsumeQueue
ConsumeQueue consumeQueue = findConsumeQueue(topic, queueId);
...
//根據 offset 找到對應的 ConsumeQueue 的 MappedFile
SelectMappedBufferResult bufferConsumeQueue = consumeQueue.getIndexBuffer(offset);
status = GetMessageStatus.NO_MATCHED_MESSAGE;
long maxPhyOffsetPulling = 0;
int i = 0;
//能返回的最大信息大小，不能大於 16M
final int maxFilterMessageCount = Math.max(16000, maxMsgNums * ConsumeQueue.CQ_STORE_UNIT_SIZE);
for (; i < bufferConsumeQueue.getSize() && i < maxFilterMessageCount; i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
//CommitLog 物理地址
long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();
int sizePy = bufferConsumeQueue.getByteBuffer().getInt();
maxPhyOffsetPulling = offsetPy;
...
//根據 offset 和 size 從 CommitLog 拿到具體的 Message
SelectMappedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy);
...
//將 Message 放入結果集
getResult.addMessage(selectResult);
status = GetMessageStatus.FOUND;
}
//更新 offset
nextBeginOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
long diff = maxOffsetPy - maxPhyOffsetPulling;
long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE
* (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));
getResult.setSuggestPullingFromSlave(diff > memory);
...
getResult.setStatus(status);
getResult.setNextBeginOffset(nextBeginOffset);
return getResult;
}
4.2 根據 key 查詢
讀取消息的過程就是用 topic 和 key 找到 IndexFile 索引文件中的一條記錄，根據記錄中的 CommitLog 的 offset 從 CommitLog 文件中讀取消息的實體內容。
//org.apache.rocketmq.store.DefaultMessageStore::queryMessage
public QueryMessageResult queryMessage(String topic, String key, int maxNum, long begin, long end) {
QueryMessageResult queryMessageResult = new QueryMessageResult();
long lastQueryMsgTime = end;
for (int i = 0; i < 3; i++) {
//獲取 IndexFile 索引文件中記錄的消息在 CommitLog 文件物理偏移地址
QueryOffsetResult queryOffsetResult = this.indexService.queryOffset(topic, key, maxNum, begin, lastQueryMsgTime);
...
for (int m = 0; m < queryOffsetResult.getPhyOffsets().size(); m++) {
long offset = queryOffsetResult.getPhyOffsets().get(m);
...
MessageExt msg = this.lookMessageByOffset(offset);
if (0 == m) {
lastQueryMsgTime = msg.getStoreTimestamp();
}
...
//在 CommitLog 文件獲取消息內容
SelectMappedBufferResult result = this.commitLog.getData(offset, false);
...
queryMessageResult.addMessage(result);
...
}
}
return queryMessageResult;
}
在 IndexFile 索引文件，查找 CommitLog 文件物理偏移地址實現如下：
//org.apache.rocketmq.store.index.IndexFile::selectPhyOffset
public void selectPhyOffset(final List<Long> phyOffsets, final String key, final int maxNum,
final long begin, final long end, boolean lock) {
int keyHash = indexKeyHashMethod(key);
int slotPos = keyHash % this.hashSlotNum;
int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
//獲取相同 hash 值 key 的第一個 IndexItme 存儲位置，即鏈表的首節點
int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
//遍歷鏈表節點
for (int nextIndexToRead = slotValue; ; ) {
if (phyOffsets.size() >= maxNum) {
break;
}
int absIndexPos =
IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
+ nextIndexToRead * indexSize;
int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);
long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);
long timeDiff = (long) this.mappedByteBuffer.getInt(absIndexPos + 4 + 8);
int prevIndexRead = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8 + 4);
if (timeDiff < 0) {
break;
}
timeDiff *= 1000L;
long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;
boolean timeMatched = (timeRead >= begin) && (timeRead <= end);
//符合條件的結果加入 phyOffsets
if (keyHash == keyHashRead && timeMatched) {
phyOffsets.add(phyOffsetRead);
}
if (prevIndexRead <= invalidIndex
|| prevIndexRead > this.indexHeader.getIndexCount()
|| prevIndexRead == nextIndexToRead || timeRead < begin) {
break;
}
//繼續遍歷鏈表
nextIndexToRead = prevIndexRead;
}
...
}
五、總結
本文從源碼的角度介紹了 RocketMQ 存儲系統的核心模塊實現，包括存儲架構、消息寫入和消息讀取。
RocketMQ 把所有 Topic 下的消息都寫入到 CommitLog 里面，實現了嚴格的順序寫。通過文件預熱防止 Page Cache 被交換到 swap 空間，減少讀寫文件時缺頁中斷。使用 mmap 對 CommitLog 文件進行讀寫，將對文件的操作轉化為直接對內存地址進行操作，從而極大地提高了文件的讀寫效率。
對於性能要求高、數據一致性要求不高的場景下，可以通過開啟堆外內存，實現讀寫分離，提升磁盤的吞吐量。總之，存儲模塊的學習需要對操作系統原理有一定了解。作者采用的性能極致優化方案值得我們好好學習。
六、參考文獻
作者：vivo互聯網服務器團隊-Zhang Zhenglin