三分鐘白話RocketMQ系列—— 如何存儲(chǔ)消息

作者：阿丸筆記 2023-08-14 09:16:26

因?yàn)橄⒋鎯?chǔ)仍然使用本地磁盤，本地磁盤空間不足時(shí)，為保證服務(wù)穩(wěn)定性消息仍然會(huì)被強(qiáng)制清理，導(dǎo)致消息的實(shí)際保存時(shí)長(zhǎng)小于設(shè)置的保存時(shí)長(zhǎng)。建議在存儲(chǔ)成本可控的前提下，盡可能延長(zhǎng)消息存儲(chǔ)時(shí)長(zhǎng)。延長(zhǎng)消息存儲(chǔ)時(shí)長(zhǎng)，可以為緊急故障恢復(fù)、應(yīng)急問(wèn)題排查和消息回溯帶來(lái)更多的可操作空間。

我們知道RocketMQ主要分為消息生產(chǎn)、存儲(chǔ)（消息堆積）、消費(fèi) 三大塊領(lǐng)域。

那接下來(lái)，我們白話一下，RocketMQ是如何存儲(chǔ)消息的，揭秘消息存儲(chǔ)全過(guò)程。

注意，如果白話中不小心提到相關(guān)代碼配置與類名，請(qǐng)參考RocketMQ 4.9.4版本

關(guān)鍵字摘要

存儲(chǔ)模型與存儲(chǔ)類型
如何保證存儲(chǔ)消息不丟失
如何提高寫入性能
如何清理過(guò)期消息

存儲(chǔ)模型是什么？有哪些存儲(chǔ)類型？

RocketMQ使用了一種基于日志的存儲(chǔ)方式，將消息以順序?qū)懭氲姆绞阶芳拥轿募校瑥亩鴮?shí)現(xiàn)高性能的消息存儲(chǔ)和讀取。

RocketMQ的消息存儲(chǔ)方式可以分為兩個(gè)類型：CommitLog 和ConsumeQueue 。

圖片

還有一個(gè)文件類型是indexfile，主要用于控制臺(tái)消息檢索，不影響消息的寫入與消費(fèi)，我們就不展開了。

CommitLog

CommitLog文件存儲(chǔ)了Producer端寫入的消息主體內(nèi)容，它以追加寫入的方式將消息存儲(chǔ)到磁盤上的文件中。

單個(gè)文件大小默認(rèn)1G ，文件名長(zhǎng)度為20位（左邊補(bǔ)零，剩余為起始偏移量），當(dāng)文件滿了，寫入下一個(gè)文件。

比如00000000000000000000代表了第一個(gè)文件，起始偏移量為0，文件大小為1G=1073741824；當(dāng)?shù)谝粋€(gè)文件寫滿了，第二個(gè)文件為00000000001073741824，起始偏移量為1073741824，以此類推。

它的主要特點(diǎn)是：順序?qū)懀请S機(jī)讀（被ConsumeQueue讀取）。

雖然是隨機(jī)讀，但是利用package機(jī)制，可以批量地從磁盤讀取，作為cache存到內(nèi)存中，加速后續(xù)的讀取速度。

Broker單個(gè)實(shí)例下所有的隊(duì)列共用一個(gè)日志數(shù)據(jù)文件CommitLog來(lái)存儲(chǔ)。而Kafka采用的是獨(dú)立型的存儲(chǔ)結(jié)構(gòu)，每個(gè)隊(duì)列一個(gè)文件。

ConsumeQueue

ConsumeQueue文件是用于支持消息消費(fèi)的存儲(chǔ)結(jié)構(gòu)。保存了指定Topic下的隊(duì)列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。

消費(fèi)者通過(guò) 順序讀取 ConsumeQueue文件，可以快速定位到消息在CommitLog中的物理存儲(chǔ)位置，從而實(shí)現(xiàn)快速消息的拉取和消費(fèi)。

從實(shí)際物理存儲(chǔ)的角度來(lái)看，每個(gè)主題Topic下的每個(gè)隊(duì)列Queue對(duì)應(yīng)一個(gè)ConsumeQueue文件。

生產(chǎn)者端的消息是順序?qū)懭隒ommitLog，消費(fèi)者端是順序讀取ConsumeQueue。但是根據(jù)ConsumeQueue的起始物理位置偏移量offset讀取消息真實(shí)內(nèi)容，實(shí)際是隨機(jī)讀取CommitLog。實(shí)現(xiàn)了消息生產(chǎn)與消息消費(fèi)、數(shù)據(jù)存儲(chǔ)和數(shù)據(jù)索引相互分離。

怎么保證存儲(chǔ)消息不丟失？

刷盤機(jī)制

Broker在把消息寫入日志文件的過(guò)程中，如果在剛收到消息時(shí)，Broker異常宕機(jī)了，那么內(nèi)存中尚未寫入磁盤的消息就會(huì)丟失了。

因此，RocketMQ持久化消息分為兩種：同步刷盤和異步刷盤（默認(rèn)配置）。

異步刷盤是指Broker收到消息后先存儲(chǔ)到PageCache，然后立即通知Producer消息已存儲(chǔ)成功，可以繼續(xù)處理業(yè)務(wù)邏輯。此后，Broker會(huì)啟動(dòng)一個(gè)異步線程將消息持久化到磁盤。然而，如果Broker在持久化到磁盤之前發(fā)生故障，消息將會(huì)丟失。

## 刷盤策略配置
flushDiskType = ASYNC_FLUSH

注意，寫入PageCache后，應(yīng)用服務(wù)宕機(jī)消息不丟失，只有機(jī)器斷電或宕機(jī)會(huì)有少量消息丟失。

相比之下，同步刷盤的方式是在消息存儲(chǔ)到緩存后不立即通知Producer，而是等待消息被持久化到磁盤后再通知Producer。這種方式確保了消息不會(huì)丟失，但性能不如異步刷盤高。一般用于金融業(yè)務(wù)。

## 刷盤策略配置
flushDiskType = SYNC_FLUSH

在選擇刷盤方式時(shí)，需要根據(jù)業(yè)務(wù)場(chǎng)景進(jìn)行權(quán)衡。

主從同步機(jī)制

即使Broker采用同步刷盤策略，但如果刷盤完成后磁盤損壞，會(huì)導(dǎo)致所有存儲(chǔ)在磁盤上的消息丟失。

即使采用了主從復(fù)制，如果主節(jié)點(diǎn)在刷盤完成后還沒(méi)有來(lái)得及將數(shù)據(jù)同步給從節(jié)點(diǎn)就發(fā)生了磁盤故障，同樣會(huì)導(dǎo)致數(shù)據(jù)丟失。

所以我們可以配置同步機(jī)制，等待從節(jié)點(diǎn)復(fù)制完成主節(jié)點(diǎn)的消息后，才去通知Producer完成了消息存儲(chǔ)。

## 主從同步策略配置
brokerRole=SYNC_MASTER

怎么提高存儲(chǔ)寫入性能？

零拷貝技術(shù)

RocketMQ通過(guò)使用內(nèi)存映射文件（包括CommitLog、 ConsumeQueue等文件）來(lái)提高IO訪問(wèn)性能，也就是我們常說(shuō)的零拷貝技術(shù)。

Java在NIO包里，引入了sendFile（FileChannel類）和MMAP（MappedByteBuffer類）兩種實(shí)現(xiàn)方式的零拷貝技術(shù)。

主流的MQ都會(huì)使用零拷貝技術(shù)，來(lái)提升IO：

Kafka：record 的讀和寫都是基于 FileChannel。index 的讀寫則基于 MMAP。
RocketMQ：讀取數(shù)據(jù)基于 MMAP，寫入數(shù)據(jù)默認(rèn)使用 MMAP。但可以通過(guò)修改配置transientStorePoolEnable參數(shù)將其配置為使用 FileChannel。作者之所以這樣設(shè)計(jì)，是為了避免 PageCache 的鎖競(jìng)爭(zhēng)，并通過(guò)兩層架構(gòu)實(shí)現(xiàn)讀寫分離。

緩沖池寫入增強(qiáng)

在不開啟RocketMQ的內(nèi)存映射增強(qiáng)方案時(shí)，RocketMQ的讀和寫都只會(huì)簡(jiǎn)單直接使用MMAP。

但是，MappedByteBuffer也存在一些缺陷：

使用虛擬內(nèi)存，超過(guò)物理內(nèi)存會(huì)導(dǎo)致內(nèi)存交換，引起磁盤IO（可能非順序IO）速度較慢。
虛擬內(nèi)存交換是受操作系統(tǒng)控制的，所以其他進(jìn)程活動(dòng)也會(huì)觸發(fā)RocketMQ內(nèi)存映射的交換。
文件內(nèi)存映射寫入PageCache時(shí)存在鎖競(jìng)爭(zhēng)，直接寫入內(nèi)存可避免競(jìng)爭(zhēng)，在異步刷盤場(chǎng)景下速度更快。

為此，RocketMQ通過(guò)transientStorePoolEnable參數(shù)控制，對(duì)寫入進(jìn)行了優(yōu)化。

如果開啟了這個(gè)參數(shù)，會(huì)將寫入拆分為兩步，寫入緩沖區(qū) + 異步刷盤的增強(qiáng)策略。

## 刷盤策略配置
flushDiskType = ASYNC_FLUSH 
transientStorePoolEnable = true

MappedFile會(huì)提前申請(qǐng)一塊直接內(nèi)存用作緩沖區(qū)，放棄使用mmap直接寫文件。

數(shù)據(jù)先寫入緩沖區(qū)，然后異步線程每200ms（且臟數(shù)據(jù)達(dá)到16K，commitCommitLogLeastPages = 4）將緩沖區(qū)的數(shù)據(jù)commit寫入FileChannel。

再喚醒定時(shí)服務(wù)（FlushRealTimeService類）將FileChannel里的數(shù)據(jù)持久化到磁盤。flush函數(shù)和commit一樣也可以傳入一個(gè)刷盤頁(yè)數(shù)，當(dāng)臟頁(yè)數(shù)量達(dá)到16K時(shí)(flushLeastPages = 4)，會(huì)進(jìn)行刷盤操作，調(diào)用FileChannel的force將內(nèi)存中的數(shù)據(jù)持久化到磁盤。

開啟transientStorePoolEnable參數(shù)后，性能最好，但是相對(duì)來(lái)說(shuō)持久化最不可靠

如何處理消息的過(guò)期和刪除？

RocketMQ 使用存儲(chǔ)時(shí)長(zhǎng)作為消息存儲(chǔ)的依據(jù)，即每個(gè)節(jié)點(diǎn)對(duì)外承諾消息的存儲(chǔ)時(shí)長(zhǎng)。在存儲(chǔ)時(shí)長(zhǎng)范圍內(nèi)的消息都會(huì)被保留，無(wú)論消息是否被消費(fèi)；超過(guò)時(shí)長(zhǎng)限制的消息則會(huì)被清理掉。

需要注意的是，在RocketMQ中，消息存儲(chǔ)時(shí)長(zhǎng)并不能完整控制消息的實(shí)際保存時(shí)間。

因?yàn)橄⒋鎯?chǔ)仍然使用本地磁盤，本地磁盤空間不足時(shí)，為保證服務(wù)穩(wěn)定性消息仍然會(huì)被強(qiáng)制清理，導(dǎo)致消息的實(shí)際保存時(shí)長(zhǎng)小于設(shè)置的保存時(shí)長(zhǎng)。

建議在存儲(chǔ)成本可控的前提下，盡可能延長(zhǎng)消息存儲(chǔ)時(shí)長(zhǎng)。延長(zhǎng)消息存儲(chǔ)時(shí)長(zhǎng)，可以為緊急故障恢復(fù)、應(yīng)急問(wèn)題排查和消息回溯帶來(lái)更多的可操作空間。