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