深度解析RocketMQ的消息存儲整體架構

1 消息存儲

消息存儲是RocketMQ中最為復雜和最為重要的一部分，本節(jié)將分別從RocketMQ的消息存儲整體架構、PageCache與Mmap內(nèi)存映射以及RocketMQ中兩種不同的刷盤方式三方面來分別展開敘述。

1.1 消息存儲整體架構

消息存儲架構圖中主要有下面三個跟消息存儲相關的文件構成。

（1） CommitLog：消息主體以及元數(shù)據(jù)的存儲主體，存儲Producer端寫入的消息主體內(nèi)容，消息內(nèi)容不是定長的。單個文件大小默認1G ，文件名長度為20位，左邊補零，剩余為起始偏移量，比如00000000000000000000代表了第一個文件，起始偏移量為0，文件大小為1G=1073741824；當?shù)谝粋€文件寫滿了，第二個文件為00000000001073741824，起始偏移量為1073741824，以此類推。消息主要是順序?qū)懭肴罩疚募?，當文件滿了，寫入下一個文件；

（2） ConsumeQueue：消息消費隊列，引入的目的主要是提高消息消費的性能，由于RocketMQ是基于主題topic的訂閱模式，消息消費是針對主題進行的，如果要遍歷commitlog文件中根據(jù)topic檢索消息是非常低效的。Consumer即可根據(jù)ConsumeQueue來查找待消費的消息。其中，ConsumeQueue（邏輯消費隊列）作為消費消息的索引，保存了指定Topic下的隊列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。consumequeue文件可以看成是基于topic的commitlog索引文件，故consumequeue文件夾的組織方式如下：topic/queue/file三層組織結構，具體存儲路徑為：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同樣consumequeue文件采取定長設計，每一個條目共20個字節(jié)，分別為8字節(jié)的commitlog物理偏移量、4字節(jié)的消息長度、8字節(jié)tag hashcode，單個文件由30W個條目組成，可以像數(shù)組一樣隨機訪問每一個條目，每個ConsumeQueue文件大小約5.72M；

（3） IndexFile：IndexFile（索引文件）提供了一種可以通過key或時間區(qū)間來查詢消息的方法。Index文件的存儲位置是： {fileName}，文件名fileName是以創(chuàng)建時的時間戳命名的，固定的單個IndexFile文件大小約為400M，一個IndexFile可以保存 2000W個索引，IndexFile的底層存儲設計為在文件系統(tǒng)中實現(xiàn)HashMap結構，故rocketmq的索引文件其底層實現(xiàn)為hash索引。

在上面的RocketMQ的消息存儲整體架構圖中可以看出，RocketMQ采用的是混合型的存儲結構，即為Broker單個實例下所有的隊列共用一個日志數(shù)據(jù)文件（即為CommitLog）來存儲。RocketMQ的混合型存儲結構（多個Topic的消息實體內(nèi)容都存儲于一個CommitLog中）針對Producer和Consumer分別采用了數(shù)據(jù)和索引部分相分離的存儲結構，Producer發(fā)送消息至Broker端，然后Broker端使用同步或者異步的方式對消息刷盤持久化，保存至CommitLog中。只要消息被刷盤持久化至磁盤文件CommitLog中，那么Producer發(fā)送的消息就不會丟失。正因為如此，Consumer也就肯定有機會去消費這條消息。當無法拉取到消息后，可以等下一次消息拉取，同時服務端也支持長輪詢模式，如果一個消息拉取請求未拉取到消息，Broker允許等待30s的時間，只要這段時間內(nèi)有新消息到達，將直接返回給消費端。這里，RocketMQ的具體做法是，使用Broker端的后臺服務線程—ReputMessageService不停地分發(fā)請求并異步構建ConsumeQueue（邏輯消費隊列）和IndexFile（索引文件）數(shù)據(jù)。

1.2 頁緩存與內(nèi)存映射

頁緩存（PageCache）是OS對文件的緩存，用于加速對文件的讀寫。一般來說，程序?qū)ξ募M行順序讀寫的速度幾乎接近于內(nèi)存的讀寫速度，主要原因就是由于OS使用PageCache機制對讀寫訪問操作進行了性能優(yōu)化，將一部分的內(nèi)存用作PageCache。對于數(shù)據(jù)的寫入，OS會先寫入至Cache內(nèi)，隨后通過異步的方式由pdflush內(nèi)核線程將Cache內(nèi)的數(shù)據(jù)刷盤至物理磁盤上。對于數(shù)據(jù)的讀取，如果一次讀取文件時出現(xiàn)未命中PageCache的情況，OS從物理磁盤上訪問讀取文件的同時，會順序?qū)ζ渌噜弶K的數(shù)據(jù)文件進行預讀取。

在RocketMQ中，ConsumeQueue邏輯消費隊列存儲的數(shù)據(jù)較少，并且是順序讀取，在page cache機制的預讀取作用下，Consume Queue文件的讀性能幾乎接近讀內(nèi)存，即使在有消息堆積情況下也不會影響性能。而對于CommitLog消息存儲的日志數(shù)據(jù)文件來說，讀取消息內(nèi)容時候會產(chǎn)生較多的隨機訪問讀取，嚴重影響性能。如果選擇合適的系統(tǒng)IO調(diào)度算法，比如設置調(diào)度算法為“Deadline”（此時塊存儲采用SSD的話），隨機讀的性能也會有所提升。

另外，RocketMQ主要通過MappedByteBuffer對文件進行讀寫操作。其中，利用了NIO中的FileChannel模型將磁盤上的物理文件直接映射到用戶態(tài)的內(nèi)存地址中（這種Mmap的方式減少了傳統(tǒng)IO將磁盤文件數(shù)據(jù)在操作系統(tǒng)內(nèi)核地址空間的緩沖區(qū)和用戶應用程序地址空間的緩沖區(qū)之間來回進行拷貝的性能開銷），將對文件的操作轉(zhuǎn)化為直接對內(nèi)存地址進行操作，從而極大地提高了文件的讀寫效率（正因為需要使用內(nèi)存映射機制，故RocketMQ的文件存儲都使用定長結構來存儲，方便一次將整個文件映射至內(nèi)存）。

1.3 消息刷盤

（1） 同步刷盤：如上圖所示，只有在消息真正持久化至磁盤后RocketMQ的Broker端才會真正返回給Producer端一個成功的ACK響應。同步刷盤對MQ消息可靠性來說是一種不錯的保障，但是性能上會有較大影響，一般適用于金融業(yè)務應用該模式較多。

（2） 異步刷盤：能夠充分利用OS的PageCache的優(yōu)勢，只要消息寫入PageCache即可將成功的ACK返回給Producer端。消息刷盤采用后臺異步線程提交的方式進行，降低了讀寫延遲，提高了MQ的性能和吞吐量。

2 通信機制

RocketMQ消息隊列集群主要包括NameServer、Broker（Master/Slave）、Producer、Consumer4個角色，基本通訊流程如下：

（1） Broker啟動后需要完成一次將自己注冊至NameServer的操作；隨后每隔30s時間定時向NameServer上報Topic路由信息。

（2） 消息生產(chǎn)者Producer作為客戶端發(fā)送消息時候，需要根據(jù)消息的Topic從本地緩存的TopicPublishInfoTable獲取路由信息。如果沒有則更新路由信息會從NameServer上重新拉取，同時Producer會默認每隔30s向NameServer拉取一次路由信息。

（3） 消息生產(chǎn)者Producer根據(jù)2）中獲取的路由信息選擇一個隊列（MessageQueue）進行消息發(fā)送；Broker作為消息的接收者接收消息并落盤存儲。

（4） 消息消費者Consumer根據(jù)2）中獲取的路由信息，并再完成客戶端的負載均衡后，選擇其中的某一個或者某幾個消息隊列來拉取消息并進行消費。

從上面1）~3）中可以看出在消息生產(chǎn)者， Broker和NameServer之間都會發(fā)生通信（這里只說了MQ的部分通信），因此如何設計一個良好的網(wǎng)絡通信模塊在MQ中至關重要，它將決定RocketMQ集群整體的消息傳輸能力與最終的性能。

rocketmq-remoting 模塊是 RocketMQ消息隊列中負責網(wǎng)絡通信的模塊，它幾乎被其他所有需要網(wǎng)絡通信的模塊（諸如rocketmq-client、rocketmq-broker、rocketmq-namesrv）所依賴和引用。為了實現(xiàn)客戶端與服務器之間高效的數(shù)據(jù)請求與接收，RocketMQ消息隊列自定義了通信協(xié)議并在Netty的基礎之上擴展了通信模塊。

2.1 Remoting通信類結構

2.2 協(xié)議設計與編解碼

在Client和Server之間完成一次消息發(fā)送時，需要對發(fā)送的消息進行一個協(xié)議約定，因此就有必要自定義RocketMQ的消息協(xié)議。同時，為了高效地在網(wǎng)絡中傳輸消息和對收到的消息讀取，就需要對消息進行編解碼。在RocketMQ中，RemotingCommand這個類在消息傳輸過程中對所有數(shù)據(jù)內(nèi)容的封裝，不但包含了所有的數(shù)據(jù)結構，還包含了編碼解碼操作。

Header字段類型Request說明Response說明

codeint請求操作碼，應答方根據(jù)不同的請求碼進行不同的業(yè)務處理應答響應碼。0表示成功，非0則表示各種錯誤

languageLanguageCode請求方實現(xiàn)的語言應答方實現(xiàn)的語言

versionint請求方程序的版本應答方程序的版本

opaqueint相當于requestId，在同一個連接上的不同請求標識碼，與響應消息中的相對應應答不做修改直接返回

flagint區(qū)分是普通RPC還是onewayRPC得標志區(qū)分是普通RPC還是onewayRPC得標志

remarkString傳輸自定義文本信息傳輸自定義文本信息

extFieldsHashMap《String， String》請求自定義擴展信息響應自定義擴展信息

可見傳輸內(nèi)容主要可以分為以下4部分：

（1） 消息長度：總長度，四個字節(jié)存儲，占用一個int類型；

（2） 序列化類型&消息頭長度：同樣占用一個int類型，第一個字節(jié)表示序列化類型，后面三個字節(jié)表示消息頭長度；

（3） 消息頭數(shù)據(jù)：經(jīng)過序列化后的消息頭數(shù)據(jù)；

（4） 消息主體數(shù)據(jù)：消息主體的二進制字節(jié)數(shù)據(jù)內(nèi)容；

2.3 消息的通信方式和流程

在RocketMQ消息隊列中支持通信的方式主要有同步（sync）、異步（async）、單向（oneway） 三種。其中“單向”通信模式相對簡單，一般用在發(fā)送心跳包場景下，無需關注其Response。這里，主要介紹RocketMQ的異步通信流程。

2.4 Reactor多線程設計

RocketMQ的RPC通信采用Netty組件作為底層通信庫，同樣也遵循了Reactor多線程模型，同時又在這之上做了一些擴展和優(yōu)化。

上面的框圖中可以大致了解RocketMQ中NettyRemotingServer的Reactor 多線程模型。一個 Reactor 主線程（eventLoopGroupBoss，即為上面的1）負責監(jiān)聽 TCP網(wǎng)絡連接請求，建立好連接，創(chuàng)建SocketChannel，并注冊到selector上。RocketMQ的源碼中會自動根據(jù)OS的類型選擇NIO和Epoll，也可以通過參數(shù)配置），然后監(jiān)聽真正的網(wǎng)絡數(shù)據(jù)。拿到網(wǎng)絡數(shù)據(jù)后，再丟給Worker線程池（eventLoopGroupSelector，即為上面的“N”，源碼中默認設置為3），在真正執(zhí)行業(yè)務邏輯之前需要進行SSL驗證、編解碼、空閑檢查、網(wǎng)絡連接管理，這些工作交給defaultEventExecutorGroup（即為上面的“M1”，源碼中默認設置為8）去做。而處理業(yè)務操作放在業(yè)務線程池中執(zhí)行，根據(jù) RomotingCommand 的業(yè)務請求碼code去processorTable這個本地緩存變量中找到對應的 processor，然后封裝成task任務后，提交給對應的業(yè)務processor處理線程池來執(zhí)行（sendMessageExecutor，以發(fā)送消息為例，即為上面的 “M2”）。從入口到業(yè)務邏輯的幾個步驟中線程池一直再增加，這跟每一步邏輯復雜性相關，越復雜，需要的并發(fā)通道越寬。

線程數(shù)線程名線程具體說明

1NettyBoss_%dReactor 主線程

NNettyServerEPOLLSelector_%d_%dReactor 線程池

M1NettyServerCodecThread_%dWorker線程池

M2RemotingExecutorThread_%d業(yè)務processor處理線程池

3 消息過濾

RocketMQ分布式消息隊列的消息過濾方式有別于其它MQ中間件，是在Consumer端訂閱消息時再做消息過濾的。RocketMQ這么做是在于其Producer端寫入消息和Consumer端訂閱消息采用分離存儲的機制來實現(xiàn)的，Consumer端訂閱消息是需要通過ConsumeQueue這個消息消費的邏輯隊列拿到一個索引，然后再從CommitLog里面讀取真正的消息實體內(nèi)容，所以說到底也是還繞不開其存儲結構。其ConsumeQueue的存儲結構如下，可以看到其中有8個字節(jié)存儲的Message Tag的哈希值，基于Tag的消息過濾正式基于這個字段值的。

主要支持如下2種的過濾方式 （1） Tag過濾方式：Consumer端在訂閱消息時除了指定Topic還可以指定TAG，如果一個消息有多個TAG，可以用||分隔。其中，Consumer端會將這個訂閱請求構建成一個 SubscriptionData，發(fā)送一個Pull消息的請求給Broker端。Broker端從RocketMQ的文件存儲層—Store讀取數(shù)據(jù)之前，會用這些數(shù)據(jù)先構建一個MessageFilter，然后傳給Store。Store從 ConsumeQueue讀取到一條記錄后，會用它記錄的消息tag hash值去做過濾，由于在服務端只是根據(jù)hashcode進行判斷，無法精確對tag原始字符串進行過濾，故在消息消費端拉取到消息后，還需要對消息的原始tag字符串進行比對，如果不同，則丟棄該消息，不進行消息消費。

（2） SQL92的過濾方式：這種方式的大致做法和上面的Tag過濾方式一樣，只是在Store層的具體過濾過程不太一樣，真正的 SQL expression 的構建和執(zhí)行由rocketmq-filter模塊負責的。每次過濾都去執(zhí)行SQL表達式會影響效率，所以RocketMQ使用了BloomFilter避免了每次都去執(zhí)行。SQL92的表達式上下文為消息的屬性。

4 負載均衡

RocketMQ中的負載均衡都在Client端完成，具體來說的話，主要可以分為Producer端發(fā)送消息時候的負載均衡和Consumer端訂閱消息的負載均衡。

4.1 Producer的負載均衡

Producer端在發(fā)送消息的時候，會先根據(jù)Topic找到指定的TopicPublishInfo，在獲取了TopicPublishInfo路由信息后，RocketMQ的客戶端在默認方式下selectOneMessageQueue（）方法會從TopicPublishInfo中的messageQueueList中選擇一個隊列（MessageQueue）進行發(fā)送消息。具體的容錯策略均在MQFaultStrategy這個類中定義。這里有一個sendLatencyFaultEnable開關變量，如果開啟，在隨機遞增取模的基礎上，再過濾掉not available的Broker代理。所謂的“l(fā)atencyFaultTolerance”，是指對之前失敗的，按一定的時間做退避。例如，如果上次請求的latency超過550Lms，就退避3000Lms；超過1000L，就退避60000L；如果關閉，采用隨機遞增取模的方式選擇一個隊列（MessageQueue）來發(fā)送消息，latencyFaultTolerance機制是實現(xiàn)消息發(fā)送高可用的核心關鍵所在。

4.2 Consumer的負載均衡

在RocketMQ中，Consumer端的兩種消費模式（Push/Pull）都是基于拉模式來獲取消息的，而在Push模式只是對pull模式的一種封裝，其本質(zhì)實現(xiàn)為消息拉取線程在從服務器拉取到一批消息后，然后提交到消息消費線程池后，又“馬不停蹄”的繼續(xù)向服務器再次嘗試拉取消息。如果未拉取到消息，則延遲一下又繼續(xù)拉取。在兩種基于拉模式的消費方式（Push/Pull）中，均需要Consumer端在知道從Broker端的哪一個消息隊列—隊列中去獲取消息。因此，有必要在Consumer端來做負載均衡，即Broker端中多個MessageQueue分配給同一個ConsumerGroup中的哪些Consumer消費。

1、Consumer端的心跳包發(fā)送

在Consumer啟動后，它就會通過定時任務不斷地向RocketMQ集群中的所有Broker實例發(fā)送心跳包（其中包含了，消息消費分組名稱、訂閱關系集合、消息通信模式和客戶端id的值等信息）。Broker端在收到Consumer的心跳消息后，會將它維護在ConsumerManager的本地緩存變量—consumerTable，同時并將封裝后的客戶端網(wǎng)絡通道信息保存在本地緩存變量—channelInfoTable中，為之后做Consumer端的負載均衡提供可以依據(jù)的元數(shù)據(jù)信息。

2、Consumer端實現(xiàn)負載均衡的核心類—RebalanceImpl

在Consumer實例的啟動流程中的啟動MQClientInstance實例部分，會完成負載均衡服務線程—RebalanceService的啟動（每隔20s執(zhí)行一次）。通過查看源碼可以發(fā)現(xiàn)，RebalanceService線程的run（）方法最終調(diào)用的是RebalanceImpl類的rebalanceByTopic（）方法，該方法是實現(xiàn)Consumer端負載均衡的核心。這里，rebalanceByTopic（）方法會根據(jù)消費者通信類型為“廣播模式”還是“集群模式”做不同的邏輯處理。這里主要來看下集群模式下的主要處理流程：

（1） 從rebalanceImpl實例的本地緩存變量—topicSubscribeInfoTable中，獲取該Topic主題下的消息消費隊列集合（mqSet）；

（2） 根據(jù)topic和consumerGroup為參數(shù)調(diào)用mQClientFactory.findConsumerIdList（）方法向Broker端發(fā)送獲取該消費組下消費者Id列表的RPC通信請求（Broker端基于前面Consumer端上報的心跳包數(shù)據(jù)而構建的consumerTable做出響應返回，業(yè)務請求碼：GET_CONSUMER_LIST_BY_GROUP）；

（3） 先對Topic下的消息消費隊列、消費者Id排序，然后用消息隊列分配策略算法（默認為：消息隊列的平均分配算法），計算出待拉取的消息隊列。這里的平均分配算法，類似于分頁的算法，將所有MessageQueue排好序類似于記錄，將所有消費端Consumer排好序類似頁數(shù)，并求出每一頁需要包含的平均size和每個頁面記錄的范圍range，最后遍歷整個range而計算出當前Consumer端應該分配到的記錄（這里即為：MessageQueue）。

（4） 然后，調(diào)用updateProcessQueueTableInRebalance（）方法，具體的做法是，先將分配到的消息隊列集合（mqSet）與processQueueTable做一個過濾比對。

上圖中processQueueTable標注的紅色部分，表示與分配到的消息隊列集合mqSet互不包含。將這些隊列設置Dropped屬性為true，然后查看這些隊列是否可以移除出processQueueTable緩存變量，這里具體執(zhí)行removeUnnecessaryMessageQueue（）方法，即每隔1s 查看是否可以獲取當前消費處理隊列的鎖，拿到的話返回true。如果等待1s后，仍然拿不到當前消費處理隊列的鎖則返回false。如果返回true，則從processQueueTable緩存變量中移除對應的Entry；

上圖中processQueueTable的綠色部分，表示與分配到的消息隊列集合mqSet的交集。判斷該ProcessQueue是否已經(jīng)過期了，在Pull模式的不用管，如果是Push模式的，設置Dropped屬性為true，并且調(diào)用removeUnnecessaryMessageQueue（）方法，像上面一樣嘗試移除Entry；

最后，為過濾后的消息隊列集合（mqSet）中的每個MessageQueue創(chuàng)建一個ProcessQueue對象并存入RebalanceImpl的processQueueTable隊列中（其中調(diào)用RebalanceImpl實例的computePullFromWhere（MessageQueue mq）方法獲取該MessageQueue對象的下一個進度消費值offset，隨后填充至接下來要創(chuàng)建的pullRequest對象屬性中），并創(chuàng)建拉取請求對象—pullRequest添加到拉取列表—pullRequestList中，最后執(zhí)行dispatchPullRequest（）方法，將Pull消息的請求對象PullRequest依次放入PullMessageService服務線程的阻塞隊列pullRequestQueue中，待該服務線程取出后向Broker端發(fā)起Pull消息的請求。其中，可以重點對比下，RebalancePushImpl和RebalancePullImpl兩個實現(xiàn)類的dispatchPullRequest（）方法不同，RebalancePullImpl類里面的該方法為空，這樣子也就回答了上一篇中最后的那道思考題了。

消息消費隊列在同一消費組不同消費者之間的負載均衡，其核心設計理念是在一個消息消費隊列在同一時間只允許被同一消費組內(nèi)的一個消費者消費，一個消息消費者能同時消費多個消息隊列。

5 事務消息

Apache RocketMQ在4.3.0版中已經(jīng)支持分布式事務消息，這里RocketMQ采用了2PC的思想來實現(xiàn)了提交事務消息，同時增加一個補償邏輯來處理二階段超時或者失敗的消息，如下圖所示。

5.1 RocketMQ事務消息流程概要

上圖說明了事務消息的大致方案，其中分為兩個流程：正常事務消息的發(fā)送及提交、事務消息的補償流程。

1.事務消息發(fā)送及提交：

（1） 發(fā)送消息（half消息）。

（2） 服務端響應消息寫入結果。

（3） 根據(jù)發(fā)送結果執(zhí)行本地事務（如果寫入失敗，此時half消息對業(yè)務不可見，本地邏輯不執(zhí)行）。

（4） 根據(jù)本地事務狀態(tài)執(zhí)行Commit或者Rollback（Commit操作生成消息索引，消息對消費者可見）

2.補償流程：

（1） 對沒有Commit/Rollback的事務消息（pending狀態(tài)的消息），從服務端發(fā)起一次“回查”

（2） Producer收到回查消息，檢查回查消息對應的本地事務的狀態(tài)

（3） 根據(jù)本地事務狀態(tài)，重新Commit或者Rollback

其中，補償階段用于解決消息Commit或者Rollback發(fā)生超時或者失敗的情況。

5.2 RocketMQ事務消息設計

1.事務消息在一階段對用戶不可見

在RocketMQ事務消息的主要流程中，一階段的消息如何對用戶不可見。其中，事務消息相對普通消息最大的特點就是一階段發(fā)送的消息對用戶是不可見的。那么，如何做到寫入消息但是對用戶不可見呢？RocketMQ事務消息的做法是：如果消息是half消息，將備份原消息的主題與消息消費隊列，然后改變主題為RMQ_SYS_TRANS_HALF_TOPIC。由于消費組未訂閱該主題，故消費端無法消費half類型的消息，然后RocketMQ會開啟一個定時任務，從Topic為RMQ_SYS_TRANS_HALF_TOPIC中拉取消息進行消費，根據(jù)生產(chǎn)者組獲取一個服務提供者發(fā)送回查事務狀態(tài)請求，根據(jù)事務狀態(tài)來決定是提交或回滾消息。

在RocketMQ中，消息在服務端的存儲結構如下，每條消息都會有對應的索引信息，Consumer通過ConsumeQueue這個二級索引來讀取消息實體內(nèi)容，其流程如下：

RocketMQ的具體實現(xiàn)策略是：寫入的如果事務消息，對消息的Topic和Queue等屬性進行替換，同時將原來的Topic和Queue信息存儲到消息的屬性中，正因為消息主題被替換，故消息并不會轉(zhuǎn)發(fā)到該原主題的消息消費隊列，消費者無法感知消息的存在，不會消費。其實改變消息主題是RocketMQ的常用“套路”，回想一下延時消息的實現(xiàn)機制。

2.Commit和Rollback操作以及Op消息的引入

在完成一階段寫入一條對用戶不可見的消息后，二階段如果是Commit操作，則需要讓消息對用戶可見；如果是Rollback則需要撤銷一階段的消息。先說Rollback的情況。對于Rollback，本身一階段的消息對用戶是不可見的，其實不需要真正撤銷消息（實際上RocketMQ也無法去真正的刪除一條消息，因為是順序?qū)懳募模５菂^(qū)別于這條消息沒有確定狀態(tài)（Pending狀態(tài)，事務懸而未決），需要一個操作來標識這條消息的最終狀態(tài)。RocketMQ事務消息方案中引入了Op消息的概念，用Op消息標識事務消息已經(jīng)確定的狀態(tài)（Commit或者Rollback）。如果一條事務消息沒有對應的Op消息，說明這個事務的狀態(tài)還無法確定（可能是二階段失敗了）。引入Op消息后，事務消息無論是Commit或者Rollback都會記錄一個Op操作。Commit相對于Rollback只是在寫入Op消息前創(chuàng)建Half消息的索引。

3.Op消息的存儲和對應關系

RocketMQ將Op消息寫入到全局一個特定的Topic中通過源碼中的方法—TransactionalMessageUtil.buildOpTopic（）；這個Topic是一個內(nèi)部的Topic（像Half消息的Topic一樣），不會被用戶消費。Op消息的內(nèi)容為對應的Half消息的存儲的Offset，這樣通過Op消息能索引到Half消息進行后續(xù)的回查操作。

4.Half消息的索引構建

在執(zhí)行二階段Commit操作時，需要構建出Half消息的索引。一階段的Half消息由于是寫到一個特殊的Topic，所以二階段構建索引時需要讀取出Half消息，并將Topic和Queue替換成真正的目標的Topic和Queue，之后通過一次普通消息的寫入操作來生成一條對用戶可見的消息。所以RocketMQ事務消息二階段其實是利用了一階段存儲的消息的內(nèi)容，在二階段時恢復出一條完整的普通消息，然后走一遍消息寫入流程。

5.如何處理二階段失敗的消息？

如果在RocketMQ事務消息的二階段過程中失敗了，例如在做Commit操作時，出現(xiàn)網(wǎng)絡問題導致Commit失敗，那么需要通過一定的策略使這條消息最終被Commit。RocketMQ采用了一種補償機制，稱為“回查”。Broker端對未確定狀態(tài)的消息發(fā)起回查，將消息發(fā)送到對應的Producer端（同一個Group的Producer），由Producer根據(jù)消息來檢查本地事務的狀態(tài)，進而執(zhí)行Commit或者Rollback。Broker端通過對比Half消息和Op消息進行事務消息的回查并且推進CheckPoint（記錄那些事務消息的狀態(tài)是確定的）。

值得注意的是，rocketmq并不會無休止的的信息事務狀態(tài)回查，默認回查15次，如果15次回查還是無法得知事務狀態(tài)，rocketmq默認回滾該消息。

6 消息查詢

RocketMQ支持按照下面兩種維度（“按照Message Id查詢消息”、“按照Message Key查詢消息”）進行消息查詢。

6.1 按照MessageId查詢消息

RocketMQ中的MessageId的長度總共有16字節(jié)，其中包含了消息存儲主機地址（IP地址和端口），消息Commit Log offset?！鞍凑誐essageId查詢消息”在RocketMQ中具體做法是：Client端從MessageId中解析出Broker的地址（IP地址和端口）和Commit Log的偏移地址后封裝成一個RPC請求后通過Remoting通信層發(fā)送（業(yè)務請求碼：VIEW_MESSAGE_BY_ID）。Broker端走的是QueryMessageProcessor，讀取消息的過程用其中的 commitLog offset 和 size 去 commitLog 中找到真正的記錄并解析成一個完整的消息返回。

6.2 按照Message Key查詢消息

“按照Message Key查詢消息”，主要是基于RocketMQ的IndexFile索引文件來實現(xiàn)的。RocketMQ的索引文件邏輯結構，類似JDK中HashMap的實現(xiàn)。索引文件的具體結構如下：

IndexFile索引文件為用戶提供通過“按照Message Key查詢消息”的消息索引查詢服務，IndexFile文件的存儲位置是： {fileName}，文件名fileName是以創(chuàng)建時的時間戳命名的，文件大小是固定的，等于40+500W*4+2000W*20= 420000040個字節(jié)大小。如果消息的properties中設置了UNIQ_KEY這個屬性，就用 topic + “#” + UNIQ_KEY的value作為 key 來做寫入操作。如果消息設置了KEYS屬性（多個KEY以空格分隔），也會用 topic + “#” + KEY 來做索引。

其中的索引數(shù)據(jù)包含了Key Hash/CommitLog Offset/Timestamp/NextIndex offset 這四個字段，一共20 Byte。NextIndex offset 即前面讀出來的 slotValue，如果有 hash沖突，就可以用這個字段將所有沖突的索引用鏈表的方式串起來了。Timestamp記錄的是消息storeTimestamp之間的差，并不是一個絕對的時間。整個Index File的結構如圖，40 Byte 的Header用于保存一些總的統(tǒng)計信息，4*500W的 Slot Table并不保存真正的索引數(shù)據(jù)，而是保存每個槽位對應的單向鏈表的頭。20*2000W 是真正的索引數(shù)據(jù)，即一個 Index File 可以保存 2000W個索引。

“按照Message Key查詢消息”的方式，RocketMQ的具體做法是，主要通過Broker端的QueryMessageProcessor業(yè)務處理器來查詢，讀取消息的過程就是用topic和key找到IndexFile索引文件中的一條記錄，根據(jù)其中的commitLog offset從CommitLog文件中讀取消息的實體內(nèi)容。

審核編輯：黃飛