一丶前言

在《Rocketmq學(xué)習(xí)3——消息發(fā)送原理源碼淺析》中，我們學(xué)習(xí)了消息發(fā)送的要點(diǎn)：

• 本地緩存+rpc 請(qǐng)求namesever + 定時(shí)刷新，topic路由信息
• 負(fù)載均衡的選擇一個(gè)Broker進(jìn)行發(fā)送，還支持【故障轉(zhuǎn)移（即支持規(guī)避短時(shí)間內(nèi)發(fā)送失敗的broker）】
• 基于netty實(shí)現(xiàn)的rpc進(jìn)行消息發(fā)送

這一篇我們將學(xué)習(xí)，消息是如何持久化在broker上的

二丶概述

消息存儲(chǔ)的流程如下：

1. 發(fā)送消息： 生產(chǎn)者（Producer）發(fā)送消息到 Broker。
2. 消息存儲(chǔ)：Broker 接收到消息后，將消息存儲(chǔ)在消息存儲(chǔ)文件中，通常是 CommitLog 文件。 RocketMQ 使用了內(nèi)存映射文件（MappedByteBuffer）來(lái)提高文件的讀寫(xiě)速度，它可以將文件直接映射到虛擬內(nèi)存，減少了文件 I/O 操作。
3. 寫(xiě)入磁盤(pán)：RocketMQ 使用了順序?qū)懙姆绞綄⑾?xiě)入到 CommitLog，這是因?yàn)轫樞驅(qū)懘疟P(pán)的速度遠(yuǎn)快于隨機(jī)寫(xiě)。
4. 索引文件更新：為了提高查詢效率，消息會(huì)被索引，索引信息存儲(chǔ)在 ConsumerQueue 和 IndexFile 中。ConsumerQueue 存儲(chǔ)了消息在 CommitLog 中的偏移量，而 IndexFile 存儲(chǔ)了關(guān)鍵字到消息偏移量的映射。這一步和broker處理消息發(fā)送請(qǐng)求是異步的，由后臺(tái)線程定時(shí)處理。
5. 數(shù)據(jù)刷盤(pán)：RocketMQ 提供兩種消息刷盤(pán)方式：
   1. 同步刷盤(pán)和異步刷盤(pán)。同步刷盤(pán)會(huì)在消息確實(shí)寫(xiě)入磁盤(pán)后再向生產(chǎn)者確認(rèn)消息發(fā)送成功，
   2. 異步刷盤(pán)則在寫(xiě)入操作系統(tǒng) PageCache 后就確認(rèn)，依靠操作系統(tǒng)異步將數(shù)據(jù)刷寫(xiě)到磁盤(pán)。
6. HA 機(jī)制：為了保證數(shù)據(jù)的高可用性，RocketMQ 還提供了主從同步機(jī)制，從服務(wù)器可以從主服務(wù)器上復(fù)制數(shù)據(jù)，確保在主服務(wù)器宕機(jī)時(shí)，從服務(wù)器可以接管消息服務(wù)。

三丶broker是接收消息發(fā)送請(qǐng)求

broker在啟動(dòng)的時(shí)候，會(huì)啟動(dòng)BrokerController,BrokerController會(huì)觸發(fā)remotingServer的啟動(dòng)。remotingServer基于netty實(shí)現(xiàn)，其中關(guān)聯(lián)了RequestCode（rocketmq協(xié)議中使用一個(gè)int表明請(qǐng)求類型）和對(duì)應(yīng)的請(qǐng)求處理的processor。

其中SEND_MESSAGE對(duì)應(yīng)的processor——SendMessageProcessor。

在broker啟動(dòng)時(shí)，會(huì)觸發(fā)基于netty的服務(wù)端啟動(dòng)，其中注冊(cè)的NettyServerHandler實(shí)現(xiàn)了ChannelInboundHandler，在數(shù)據(jù)客戶端數(shù)據(jù)到達(dá)的時(shí)候會(huì)先經(jīng)由解碼器ByteToMessageDecoder（rocketmq根據(jù)自己的協(xié)議實(shí)現(xiàn)了解碼器——NettyDecoder），解碼后將調(diào)用到如下的NettyServerHandler！

其中會(huì)根據(jù)請(qǐng)求類型，獲取到對(duì)應(yīng)的Processor，消息發(fā)送一般最后由SendMessageProcessor處理

四丶rocketmq基于netty實(shí)現(xiàn)的遠(yuǎn)程服務(wù)處理請(qǐng)求的流程

SendMessageProcessor接收到請(qǐng)求的時(shí)候，不是立馬在當(dāng)前線程進(jìn)行處理，而是將封裝成一個(gè)任務(wù)，提交到業(yè)務(wù)線程池。

在提交之前，還是會(huì)進(jìn)行當(dāng)前broker是否關(guān)閉中，是否拒絕請(qǐng)求的判斷。

如下是處理請(qǐng)求的大致流程

可看到綠色部分才是真正處理請(qǐng)求的部分，處理后將響應(yīng)寫(xiě)到netty的channel中，實(shí)習(xí)響應(yīng)！

五丶SendMessageProcessor 處理請(qǐng)求大致流程

rocketmq留了一堆擴(kuò)展的鉤子，最終在sendMessage方法中進(jìn)行一系列的校驗(yàn)，包裝消息為MessageExtBrokerInner，然后進(jìn)行消息存儲(chǔ)流程，源碼如下

消息存儲(chǔ)最后交給MessageStore，調(diào)用asyncPutMessage進(jìn)行異步存儲(chǔ)消息，也就是說(shuō)業(yè)務(wù)處理線程并沒(méi)有一直阻塞到消息存儲(chǔ)完畢，而是提交后就釋放了

看到這里你可能會(huì)疑問(wèn)，那么同步消息發(fā)送者豈不是收不到響應(yīng)，同步消息消費(fèi)者還會(huì)block住么？

還是會(huì)的，因?yàn)?/p>

只有在MessageStore異步存儲(chǔ)完消息后，才會(huì)回調(diào)doResponse寫(xiě)回響應(yīng)！

這樣做的目的在于將業(yè)務(wù)處理Executor，和消息存儲(chǔ)Executor進(jìn)行解耦

六丶消息持久化

可看到最終使用CommitLog進(jìn)行消息存儲(chǔ)

1.消息持久化前置流程

如上主要是進(jìn)行一些校驗(yàn)，其中有兩層鎖

• topic + queue鎖

  topicQueueKey由topic和queueId構(gòu)成，因?yàn)橐粋€(gè)broker上可有多個(gè)topic，一個(gè)topic可具備多個(gè)messgeQuque，這里使用hash實(shí)現(xiàn)鎖粒度的細(xì)化，那么queueId是在哪里生成的？

  如上是在SendMessageProcessor中，如果指定了queueId那么使用指定的queueId，反之隨機(jī)產(chǎn)生一個(gè)。

• 文件鎖

  rocketmq具備兩個(gè)實(shí)現(xiàn)：

  • 一個(gè)基于AQS ReentrantLock
  • 一個(gè)基于cas自旋

  高并發(fā)情況浪費(fèi)大量cpu,低并發(fā)情況下減少內(nèi)核態(tài)用戶態(tài)切換

2.消息持久化

2.1 MappedFile文件創(chuàng)建

這里會(huì)構(gòu)建出兩個(gè)文件路徑，這意味著會(huì)一次性創(chuàng)建兩個(gè)文件，可看到文件名稱是偏移量的大小——比如00000000000000000000代表了第一個(gè)文件，起始偏移量為0，文件大小為1G=1073741824；當(dāng)?shù)谝粋€(gè)文件寫(xiě)滿了，第二個(gè)文件為00000000001073741824，起始偏移量為1073741824，以此類推

下面我們看看文件創(chuàng)建的源碼：

文件創(chuàng)建并不是由當(dāng)前線程進(jìn)行的，而是將請(qǐng)求提交到requestTable中，然后等待指定時(shí)間。

然后再背后存在一個(gè)線程，不斷從隊(duì)列中拿任務(wù)進(jìn)行處理

可以看到MappedFile支持SPI機(jī)制，但是這里的代碼讓人作嘔

如果開(kāi)啟的了堆外內(nèi)存緩沖，那么會(huì)使用：new DefaultMappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool())創(chuàng)建DefaultMappedFile

否則使用new DefaultMappedFile(req.getFilePath(), req.getFileSize())創(chuàng)建DefaultMappedFile。

1. 不適應(yīng)堆外內(nèi)存緩沖

   使用fileChannel.map創(chuàng)建mappedByteBuffer

2. 使用堆外寫(xiě)緩沖

   會(huì)從TransientStorePool中獲取一個(gè)ByteBuffer

   這里堆外緩沖是TransientStorePool初始化時(shí)申請(qǐng)的

2.2 文件預(yù)熱

至此完成了文件的創(chuàng)建，rocktmq還會(huì)進(jìn)行文件的預(yù)熱：

預(yù)熱的過(guò)程其實(shí)就是每隔4K寫(xiě)入0值，這樣做的好處是：

提高文件的訪問(wèn)效率，尤其是在使用內(nèi)存映射（Memory Mapped File，MMF）技術(shù)時(shí)。內(nèi)存映射文件技術(shù)能將文件直接映射到操作系統(tǒng)的虛擬內(nèi)存中，進(jìn)而可以像訪問(wèn)內(nèi)存一樣訪問(wèn)這些文件，這樣可以顯著提高文件I/O的效率。

預(yù)熱(mappedFile)的過(guò)程，主要是提前將文件內(nèi)容加載到物理內(nèi)存中，確保在實(shí)際使用這些文件時(shí)，能夠避免或減少磁盤(pán)I/O帶來(lái)的延遲。因?yàn)楫?dāng)進(jìn)程首次訪問(wèn)內(nèi)存映射文件中的某個(gè)部分時(shí)，如果這部分?jǐn)?shù)據(jù)還沒(méi)有加載到物理內(nèi)存中，操作系統(tǒng)需要從磁盤(pán)中讀取數(shù)據(jù)到物理內(nèi)存，這個(gè)過(guò)程稱為缺頁(yè)中斷（page fault）。缺頁(yè)中斷會(huì)導(dǎo)致一定的延遲。

進(jìn)行預(yù)熱主要是通過(guò)以下幾種方式：

1. 觸摸內(nèi)存: 遍歷映射文件的每一頁(yè)并寫(xiě)入少量數(shù)據(jù)（例如0），這樣可以確保操作系統(tǒng)將這些頁(yè)加載到物理內(nèi)存中。
2. mlock: 在某些系統(tǒng)中，可以使用 mlock 或類似的調(diào)用來(lái)鎖定內(nèi)存的特定區(qū)域，確保這些區(qū)域常駐內(nèi)存，不會(huì)被操作系統(tǒng)交換到磁盤(pán)上（swap out）。

2.3 消息寫(xiě)入

寫(xiě)入的時(shí)候會(huì)獲取ByteBuffer，如下：如果具備寫(xiě)堆外內(nèi)存緩沖，那么使用堆外內(nèi)存，反之使用mmap生成的byteBuffer

最終就是將消息按照消息格式put到ByteBuffer中

2.4 消息刷盤(pán)

當(dāng)消息寫(xiě)入到ByteBuffer后，會(huì)進(jìn)行持久化 和高可用同步副本

這里我們看下刷盤(pán)的源碼

可以看到根據(jù)是否由堆外寫(xiě)緩沖和刷盤(pán)方式，會(huì)使用不同的service進(jìn)行wakeup實(shí)現(xiàn)刷盤(pán)：

• 堆外寫(xiě)緩沖（WriteBuffer）（只有異步刷盤(pán)模式才可以開(kāi)啟）

  在RocketMQ中，MappedFile類代表一個(gè)內(nèi)存映射文件，可以在構(gòu)造時(shí)選擇是否啟用“堆外寫(xiě)緩沖”（transientStorePoolEnable）。如果啟用，RocketMQ會(huì)創(chuàng)建一個(gè)堆外內(nèi)存池TransientStorePool，用于臨時(shí)存儲(chǔ)即將寫(xiě)入文件的數(shù)據(jù)。

  寫(xiě)入過(guò)程分為兩步：

  1. 寫(xiě)入堆外內(nèi)存：生產(chǎn)者發(fā)送的消息首先被寫(xiě)入到堆外內(nèi)存池中的一個(gè)緩沖區(qū)，這個(gè)緩沖區(qū)對(duì)應(yīng)一個(gè)ByteBuffer。
  2. 提交到MappedFile：隨后，數(shù)據(jù)會(huì)從堆外內(nèi)存緩沖區(qū)“提交”（commit）到MappedByteBuffer。在RocketMQ中，commit操作實(shí)際上是將堆外內(nèi)存中的數(shù)據(jù)復(fù)制到內(nèi)存映射文件的MappedByteBuffer中。

RocketMQ通過(guò)這種方式實(shí)現(xiàn)了一種內(nèi)存雙寫(xiě)的機(jī)制：先寫(xiě)入堆外內(nèi)存，然后再提交到內(nèi)存映射文件中。這樣做可以利用堆外內(nèi)存池做一層緩沖，提高寫(xiě)入效率，同時(shí)減少JVM垃圾回收的壓力。

• 刷盤(pán)方式（Flush）
  1. 同步刷盤(pán)（SYNC_FLUSH）：每次消息寫(xiě)入MappedByteBuffer之后，同步調(diào)用MappedByteBuffer.force()方法，將數(shù)據(jù)強(qiáng)制刷寫(xiě)到磁盤(pán)。同步刷盤(pán)提供了較高的數(shù)據(jù)安全性，但會(huì)犧牲一些性能。
  2. 異步刷盤(pán)（ASYNC_FLUSH）：消息寫(xiě)入MappedByteBuffer之后，并不立即刷寫(xiě)到磁盤(pán)，而是由后臺(tái)線程（如FlushRealTimeService）定期調(diào)用MappedByteBuffer.force()方法進(jìn)行刷盤(pán)。異步刷盤(pán)犧牲了部分?jǐn)?shù)據(jù)安全性，但提高了性能。

在RocketMQ中，刷盤(pán)策略可以根據(jù)數(shù)據(jù)的重要性和對(duì)性能的要求來(lái)選擇。如果數(shù)據(jù)安 全性要求極高，可以選擇同步刷盤(pán)；如果追求高吞吐量，可以選擇異步刷盤(pán)。

2.4.1 同步刷盤(pán)

同步刷盤(pán)，rocketmq的消息可以設(shè)置是否等待消息存儲(chǔ)完成，如下

1. 如果設(shè)置了等待刷盤(pán)成功，那么會(huì)向GroupCommitService中提交刷盤(pán)請(qǐng)求，然后返回對(duì)應(yīng)future
2. 如果沒(méi)有，那么喚醒刷盤(pán)線程，然后返回

GroupCommitService是為同步刷盤(pán)模式設(shè)計(jì)的，它允許在將消息持久化到磁盤(pán)之前暫停生產(chǎn)者的發(fā)送操作。這是為了確保在任何時(shí)候發(fā)生故障時(shí)消息不會(huì)丟失。

當(dāng)一個(gè)生產(chǎn)者請(qǐng)求將消息同步刷盤(pán)到磁盤(pán)時(shí)，它會(huì)創(chuàng)建一個(gè)GroupCommitRequest。這個(gè)請(qǐng)求包含了刷盤(pán)所需的信息，如期望刷盤(pán)的偏移量。然后，生產(chǎn)者線程將這個(gè)請(qǐng)求提交給GroupCommitService并等待。

GroupCommitService內(nèi)部維護(hù)了一個(gè)請(qǐng)求隊(duì)列。這個(gè)隊(duì)列是線程安全的，生產(chǎn)者通過(guò)putRequest方法將請(qǐng)求添加到隊(duì)列。添加請(qǐng)求后，生產(chǎn)者線程調(diào)用CountDownLatch.await()方法等待。

GroupCommitService的主循環(huán)會(huì)檢查隊(duì)列中是否有請(qǐng)求。如果有，它會(huì)將這些請(qǐng)求從隊(duì)列中移除并進(jìn)行處理。處理包括將CommitLog中的相關(guān)數(shù)據(jù)刷盤(pán)到磁盤(pán)。一旦完成，它將調(diào)用每個(gè)GroupCommitRequest的wakeupCustomer方法，該方法將減少CountDownLatch的計(jì)數(shù)，從而允許等待的生產(chǎn)者線程繼續(xù)執(zhí)行。

如下是GroupCommitService處理刷盤(pán)，和異步刷盤(pán)的不同在于其會(huì)設(shè)置刷盤(pán)future狀態(tài)，從而讓等待刷盤(pán)的線程被喚醒

2.4.3 異步刷盤(pán)

異步刷盤(pán)針對(duì)是否開(kāi)啟了堆外寫(xiě)緩沖會(huì)調(diào)用不同的Service

1. 開(kāi)啟了堆外寫(xiě)緩沖：使用CommitRealTimeService

   與GroupCommitService不同，CommitRealTimeService是為異步刷盤(pán)模式設(shè)計(jì)的，它不會(huì)在每次消息追加到CommitLog后暫停生產(chǎn)者線程。相反，它根據(jù)預(yù)設(shè)的時(shí)間間隔或消息積累量定期刷盤(pán)。
   

   如下是刷盤(pán)的源碼：

2. 沒(méi)開(kāi)啟堆外寫(xiě)緩沖：使用FlushRealTimeService，其會(huì)調(diào)用flush直接進(jìn)行刷新

七丶高可用

讓我們回到CommitLog#asyncPutMessage方法，可以看到下面有一個(gè)高可用的處理（needHandleHA）

那什么是否需要刷新到其他副本昵？

Message必須setWaitStoreMsgOK(true)，且消息存儲(chǔ)表明需要副本，并且角色是SYNC_MASTER

那么高可用如何實(shí)現(xiàn)的？

下面是RocketMq高可用機(jī)制：

RocketMQ 通過(guò)其 HAService（高可用性服務(wù)）實(shí)現(xiàn)了主從同步復(fù)制，確保了消息的高可用性。它的工作原理是在主Broker（Master）上的CommitLog更新之后，這些更新會(huì)被復(fù)制到一個(gè)或多個(gè)從Broker（Slave）上。這樣，即使主Broker發(fā)生故障，從Broker也可以接管工作，保證消息服務(wù)的可用性。

HAService 主要包括兩個(gè)組件：Master端的 HAService 和 Slave端的 HAConnection。

1.Master 端

HAService 主要負(fù)責(zé)管理與從Broker的連接，并將CommitLog的更新推送到所有連接的從Broker上。

1. 連接建立：HAService 在Master上監(jiān)聽(tīng)一個(gè)特定的端口。從Broker通過(guò)這個(gè)端口與Master建立連接。

2. 封裝連接：HAService 管理所有的從Broker連接。每當(dāng)有新的從Broker連接到Master時(shí)，它都會(huì)創(chuàng)建一個(gè)新的HAConnection。

3. 數(shù)據(jù)同步： Master上的CommitLog更新后，HAService 會(huì)將這些數(shù)據(jù)通過(guò)HAConnection發(fā)送到從Broker。數(shù)據(jù)的發(fā)送依賴于從Broker的拉取請(qǐng)求，即從Broker告訴Master它已經(jīng)接收了哪些數(shù)據(jù)，并請(qǐng)求后續(xù)的數(shù)據(jù)。

4. 資源清理：如果從Broker斷開(kāi)連接或出現(xiàn)錯(cuò)誤，HAService 會(huì)關(guān)閉對(duì)應(yīng)的HAConnection并清理資源。

2. Slave 端

從Broker端的HAConnection主要負(fù)責(zé)與Master保持通信，獲取數(shù)據(jù)更新，并將這些更新寫(xiě)入本地的CommitLog。

1. 連接建立： 從Broker啟動(dòng)時(shí)，它會(huì)嘗試與Master建立HAConnection。

2. 請(qǐng)求數(shù)據(jù)并同步： 從Broker會(huì)定期發(fā)送已確認(rèn)的數(shù)據(jù)偏移量給Master，并請(qǐng)求新的數(shù)據(jù)。收到Master的數(shù)據(jù)后，從Broker會(huì)將數(shù)據(jù)寫(xiě)入自己的CommitLog。

3. 反饋同步進(jìn)度：從Broker在成功將數(shù)據(jù)寫(xiě)入CommitLog后，會(huì)更新已確認(rèn)的數(shù)據(jù)偏移量，并準(zhǔn)備發(fā)送回Master以獲取更多數(shù)據(jù)。

通過(guò)這種方式，RocketMQ的HAService確保了消息數(shù)據(jù)在Master和Slave之間實(shí)時(shí)同步，即使在Master出現(xiàn)故障的情況下，也能保證服務(wù)的高可用性。

需要注意的是，這種主從同步機(jī)制雖然提供了高可用性，但它可能會(huì)對(duì)消息的發(fā)送性能產(chǎn)生一定影響，因?yàn)镸aster需要在將消息存儲(chǔ)到本地CommitLog并且同步到從Broker之后才能向生產(chǎn)者發(fā)送確認(rèn)響應(yīng)。此外，如果從Broker落后于Master太多，也有可能影響整體的同步效率。

為了確保數(shù)據(jù)的強(qiáng)一致性，RocketMQ通常建議至少部署一個(gè)Master和一個(gè)Slave，并在Broker配置中設(shè)置brokerRole為SYNC_MASTER。這樣，只有當(dāng)數(shù)據(jù)成功復(fù)制到至少一個(gè)Slave時(shí)，Master才會(huì)對(duì)消息發(fā)送者確認(rèn)成功。這確保了即使Master發(fā)生故障，消息也不會(huì)丟失，因?yàn)橹辽儆幸粋€(gè)Slave擁有完整的數(shù)據(jù)副本。

八丶總結(jié)

感覺(jué)rocketmq代碼寫(xiě)的很垃圾，但是功能還是實(shí)現(xiàn)了的。其落盤(pán)+副本同步，再很多其他中間件中也是適用的

2.mmap內(nèi)存映射文件

內(nèi)存映射文件（Memory-Mapped File，簡(jiǎn)稱 mmap）

• 內(nèi)存映射：mmap 通過(guò)將磁盤(pán)上的文件映射到虛擬內(nèi)存的方式，使得應(yīng)用程序可以像訪問(wèn)內(nèi)存一樣直接讀寫(xiě)文件區(qū)域，避免了傳統(tǒng)的文件I/O操作（read/write）中用戶空間和內(nèi)核空間之間上下文切換的開(kāi)銷。
• 操作系統(tǒng)緩存：mmap 的數(shù)據(jù)可以被操作系統(tǒng)自動(dòng)地緩存，提高了數(shù)據(jù)訪問(wèn)速度。操作系統(tǒng)會(huì)負(fù)責(zé)將修改過(guò)的內(nèi)存數(shù)據(jù)同步回文件，減少了顯式的讀寫(xiě)操作

3.文件預(yù)熱

預(yù)熱(mappedFile)的過(guò)程，主要是提前將文件內(nèi)容加載到物理內(nèi)存中，確保在實(shí)際使用這些文件時(shí)，能夠避免或減少磁盤(pán)I/O帶來(lái)的延遲。因?yàn)楫?dāng)進(jìn)程首次訪問(wèn)內(nèi)存映射文件中的某個(gè)部分時(shí)，如果這部分?jǐn)?shù)據(jù)還沒(méi)有加載到物理內(nèi)存中，操作系統(tǒng)需要從磁盤(pán)中讀取數(shù)據(jù)到物理內(nèi)存，這個(gè)過(guò)程稱為缺頁(yè)中斷（page fault）。缺頁(yè)中斷會(huì)導(dǎo)致一定的延遲。

4.堆外內(nèi)存寫(xiě)緩沖

rocketmq支持開(kāi)啟堆外寫(xiě)緩沖，優(yōu)先寫(xiě)到DirectByteBuffer中，然后使用FileChannel#write刷新到pageCache，這樣做好處是可以將多個(gè)消息先聚合到堆外byteBuffer然后一次性寫(xiě)入到page'Cache，減少系統(tǒng)調(diào)用。

5.HA機(jī)制

同步到副本，避免master宕機(jī)時(shí)消息丟失。

弊端是如果寫(xiě)master成功，同步副本失敗，消息生產(chǎn)者maybe重試，導(dǎo)致消息重復(fù)，以及同步副本帶來(lái)的延遲降低了系統(tǒng)的吞吐量。

總結(jié)

以上是生活随笔為你收集整理的Rocketmq学习4——Broker消息持久化原理源码浅析的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。