[導(dǎo)讀]作者：vivo互聯(lián)網(wǎng)服務(wù)器團(tuán)隊-ZhangZhenglin一、簡介RocketMQ是阿里巴巴開源的分布式消息中間件，它借鑒了Kafka實現(xiàn)，支持消息訂閱與發(fā)布、順序消息、事務(wù)消息、定時消息、消息回溯、死信隊列等功能。RocketMQ架構(gòu)上主要分為四部分，如下圖所示：Produce...
一、簡介
RocketMQ 是阿里巴巴開源的分布式消息中間件，它借鑒了 Kafka 實現(xiàn)，支持消息訂閱與發(fā)布、順序消息、事務(wù)消息、定時消息、消息回溯、死信隊列等功能。RocketMQ 架構(gòu)上主要分為四部分，如下圖所示：
- Producer：消息生產(chǎn)者，支持分布式集群方式部署。
- Consumer：消息消費者，支持分布式集群方式部署。
- NameServer：名字服務(wù)，是一個非常簡單的 Topic 路由注冊中心，支持 Broker 的動態(tài)注冊與發(fā)現(xiàn)，Producer 和 Consumer 通過 NameServer 動態(tài)感知 Broker 的路由信息。
- Broker：Broker 主要負(fù)責(zé)消息的存儲、轉(zhuǎn)發(fā)和查詢。
本文基于 Apache RocketMQ 4.9.1 版本剖析 Broker 中的消息存儲模塊是如何設(shè)計的。
二、存儲架構(gòu)
RocketMQ 的消息文件路徑如圖所示。
CommitLog消息主體以及元數(shù)據(jù)的存儲主體，存儲 Producer 端寫入的消息主體內(nèi)容，消息內(nèi)容不是定長的。單個文件大小默認(rèn)1G， 文件名長度為 20 位，左邊補(bǔ)零，剩余為起始偏移量，比如 00000000000000000000 代表了第一個文件，起始偏移量為 0，文件大小為 1G=1073741824；當(dāng)?shù)谝粋€文件寫滿了，第二個文件為 00000000001073741824，起始偏移量為 1073741824，以此類推。
ConsumeQueue消息消費隊列，Consumequeue 文件可以看成是基于 CommitLog 的索引文件。ConsumeQueue 文件采取定長設(shè)計，每一個條目共 20 個字節(jié)，分別為 8 字節(jié)的 CommitLog 物理偏移量、4 字節(jié)的消息長度、8 字節(jié) tag hashcode，單個文件由 30W 個條目組成，可以像數(shù)組一樣隨機(jī)訪問每一個條目，每個 ConsumeQueue 文件大小約 5.72M。
IndexFile索引文件，提供了一種可以通過 key 或時間區(qū)間來查詢消息的方法。單個 IndexFile 文件大小約為 400M，一個 IndexFile 可以保存 2000W 個索引，IndexFile 的底層存儲設(shè)計類似 JDK 的 HashMap 數(shù)據(jù)結(jié)構(gòu)。
其他文件：包括 config 文件夾，存放運行時配置信息；abort 文件，說明 Broker 是否正常關(guān)閉；checkpoint 文件，存儲 Commitlog、ConsumeQueue、Index 文件最后一次刷盤時間戳。這些不在本文討論的范圍。
同 Kafka 相比，Kafka 每個 Topic 的每個 partition 對應(yīng)一個文件，順序?qū)懭?，定時刷盤。但一旦單個 Broker 的 Topic 過多，順序?qū)憣⑼嘶癁殡S機(jī)寫。而 RocketMQ 單個 Broker 所有 Topic 在同一個 CommitLog 中順序?qū)懀悄軌虮ＷC嚴(yán)格順序?qū)?。RocketMQ 讀取消息需要從 ConsumeQueue 中拿到消息實際物理偏移再去 CommitLog 讀取消息內(nèi)容，會造成隨機(jī)讀取。
2.1 Page Cache 和 mmap
在正式介紹 Broker 消息存儲模塊實現(xiàn)前，先說明下 Page Cache 和 mmap 這兩個概念。
Page Cache 是 OS 對文件的緩存，用于加速對文件的讀寫。一般來說，程序?qū)ξ募M(jìn)行順序讀寫的速度幾乎接近于內(nèi)存的讀寫速度，主要原因就是由于 OS 使用 Page Cache 機(jī)制對讀寫訪問操作進(jìn)行了性能優(yōu)化，將一部分的內(nèi)存用作 Page Cache。對于數(shù)據(jù)的寫入，OS 會先寫入至 Cache 內(nèi)，隨后通過異步的方式由 pdflush 內(nèi)核線程將 Cache 內(nèi)的數(shù)據(jù)刷盤至物理磁盤上。對于數(shù)據(jù)的讀取，如果一次讀取文件時出現(xiàn)未命中 Page Cache 的情況，OS 從物理磁盤上訪問讀取文件的同時，會順序?qū)ζ渌噜弶K的數(shù)據(jù)文件進(jìn)行預(yù)讀取。
mmap 是將磁盤上的物理文件直接映射到用戶態(tài)的內(nèi)存地址中，減少了傳統(tǒng) IO 將磁盤文件數(shù)據(jù)在操作系統(tǒng)內(nèi)核地址空間的緩沖區(qū)和用戶應(yīng)用程序地址空間的緩沖區(qū)之間來回進(jìn)行拷貝的性能開銷。Java NIO 中的 FileChannel 提供了 map()? 方法可以實現(xiàn) mmap。FileChannel (文件通道)和 mmap (內(nèi)存映射) 讀寫性能比較可以參照這篇文章。
2.2?Broker 模塊
下圖是 Broker 存儲架構(gòu)圖，展示了 Broker 模塊從收到消息到返回響應(yīng)業(yè)務(wù)流轉(zhuǎn)過程。
業(yè)務(wù)接入層：RocketMQ 基于 Netty 的 Reactor 多線程模型實現(xiàn)了底層通信。Reactor 主線程池 eventLoopGroupBoss 負(fù)責(zé)創(chuàng)建 TCP 連接，默認(rèn)只有一個線程。連接建立后，再丟給 Reactor 子線程池 eventLoopGroupSelector 進(jìn)行讀寫事件的處理。
defaultEventExecutorGroup 負(fù)責(zé) SSL 驗證、編解碼、空閑檢查、網(wǎng)絡(luò)連接管理。然后根據(jù) RomotingCommand 的業(yè)務(wù)請求碼 code 去 processorTable 這個本地緩存變量中找到對應(yīng)的 processor，封裝成 task 任務(wù)后，提交給對應(yīng)的業(yè)務(wù) processor 處理線程池來執(zhí)行。Broker 模塊通過這四級線程池提升系統(tǒng)吞吐量。
業(yè)務(wù)處理層：處理各種通過 RPC 調(diào)用過來的業(yè)務(wù)請求，其中：
- SendMessageProcessor 負(fù)責(zé)處理 Producer 發(fā)送消息的請求；
- PullMessageProcessor 負(fù)責(zé)處理 Consumer 消費消息的請求；
- QueryMessageProcessor 負(fù)責(zé)處理按照消息 Key 等查詢消息的請求。
文件映射層：把 Commitlog、ConsumeQueue、IndexFile 文件映射為存儲對象 MappedFile。
數(shù)據(jù)傳輸層：支持基于 mmap 內(nèi)存映射進(jìn)行讀寫消息，同時也支持基于 mmap 進(jìn)行讀取消息、堆外內(nèi)存寫入消息的方式進(jìn)行讀寫消息。
下面章節(jié)將從源碼角度來剖析 RocketMQ 是如何實現(xiàn)高性能存儲。
三、消息寫入
以單個消息生產(chǎn)為例，消息寫入時序邏輯如下圖，業(yè)務(wù)邏輯如上文 Broker 存儲架構(gòu)所示在各層之間進(jìn)行流轉(zhuǎn)。
最底層消息寫入核心代碼在 CommitLog 的 asyncPutMessage 方法中，主要分為獲取 MappedFile、往緩沖區(qū)寫消息、提交刷盤請求三步。需要注意的是在這三步前后有自旋鎖或 ReentrantLock 的加鎖、釋放鎖，保證單個 Broker 寫消息是串行的。
//org.apache.rocketmq.store.CommitLog::asyncPutMessagepublic CompletableFuture asyncPutMessage(final MessageExtBrokerInner msg) { ... putMessageLock.lock(); //spin or ReentrantLock ,depending on store config try { //獲取最新的 MappedFile MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); ... //向緩沖區(qū)寫消息 result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext); ... //提交刷盤請求 CompletableFuture flushResultFuture = submitFlushRequest(result, msg); ... } finally { putMessageLock.unlock(); } ... }下面介紹這三步具體做了什么事情。
3.1?MappedFile 初始化
在 Broker 初始化時會啟動管理 MappedFile 創(chuàng)建的 AllocateMappedFileService 異步線程。消息處理線程 和 AllocateMappedFileService 線程通過隊列 requestQueue 關(guān)聯(lián)。
消息寫入時調(diào)用 AllocateMappedFileService 的 putRequestAndReturnMappedFile 方法往 requestQueue 放入提交創(chuàng)建 MappedFile 請求，這邊會同時構(gòu)建兩個 AllocateRequest 放入隊列。
AllocateMappedFileService 線程循環(huán)從 requestQueue 獲取 AllocateRequest 來創(chuàng)建 MappedFile。消息處理線程通過 CountDownLatch 等待獲取第一個 MappedFile 創(chuàng)建成功就返回。
當(dāng)消息處理線程需要再次創(chuàng)建 MappedFile 時，此時可以直接獲取之前已預(yù)創(chuàng)建的 MappedFile。這樣通過預(yù)創(chuàng)建 MappedFile ，減少文件創(chuàng)建等待時間。
//org.apache.rocketmq.store.AllocateMappedFileService::putRequestAndReturnMappedFilepublic MappedFile putRequestAndReturnMappedFile(String nextFilePath, String nextNextFilePath, int fileSize) { //請求創(chuàng)建 MappedFile AllocateRequest nextReq = new AllocateRequest(nextFilePath, fileSize); boolean nextPutOK = this.requestTable.putIfAbsent(nextFilePath, nextReq) == null; ... //請求預(yù)先創(chuàng)建下一個 MappedFile AllocateRequest nextNextReq = new AllocateRequest(nextNextFilePath, fileSize); boolean nextNextPutOK = this.requestTable.putIfAbsent(nextNextFilePath, nextNextReq) == null; ... //獲取本次創(chuàng)建 MappedFile AllocateRequest result = this.requestTable.get(nextFilePath); ...} //org.apache.rocketmq.store.AllocateMappedFileService::runpublic void run() { .. while (!this.isStopped() 
下載文檔
