RocketMQ如何保證消息的可靠性？
消息的發(fā)送方式有哪幾種?存儲(chǔ)消息的可靠性面臨哪些挑戰(zhàn)?消費(fèi)消息的確認(rèn)機(jī)制是怎樣的?本文通過分析消息流轉(zhuǎn)的整個(gè)過程，從消息發(fā)送、消息存儲(chǔ)和消息消費(fèi)三個(gè)階段介紹RocketMQ是如何保證消息的可靠性的。
分布式系統(tǒng)中一個(gè)重要的前提假設(shè)是所有的網(wǎng)絡(luò)傳輸都是不可靠的，在網(wǎng)絡(luò)傳輸不可靠的情況下，保證消息的可靠傳輸，除了進(jìn)行重試投遞別無(wú)他法。常用的絕大多數(shù)消息隊(duì)列RocketMQ、RabbitMQ等在消息傳輸上都只能保證至少傳輸成功一次，也即(At least once)，而不能保證只傳輸成功一次(Exactly once)。由于分布式系統(tǒng)網(wǎng)絡(luò)的不可靠，可能就會(huì)出現(xiàn)消息丟失的現(xiàn)象，那么RocketMQ是如何最大限度的保證消息不丟失的呢?那就需要從消息的產(chǎn)生到最終消費(fèi)的整個(gè)過程來分析，消息完整鏈路可以劃分為以下三個(gè)階段：
- 生產(chǎn)階段：消息在 Producer 發(fā)送端創(chuàng)建出來，經(jīng)過網(wǎng)絡(luò)傳輸發(fā)送到 Broker 存儲(chǔ)端。
- 存儲(chǔ)階段：消息在 Broker 端存儲(chǔ)，如果是主備或者多副本，消息會(huì)在這個(gè)階段被復(fù)制到其他的節(jié)點(diǎn)或者副本上。
- 消費(fèi)階段：Consumer 消費(fèi)端從 Broker存儲(chǔ)端拉取消息，經(jīng)過網(wǎng)絡(luò)傳輸發(fā)送到 Consumer 消費(fèi)端上，并通過重試來最大限度的保證消息的消費(fèi)。
一 發(fā)送端消息可靠性
發(fā)送端Producer發(fā)送消息Broker端的核心邏輯如下圖所示：
消息發(fā)送一般有以下幾種方式：同步發(fā)送、異步發(fā)送以及單向發(fā)送，業(yè)務(wù)具體選擇哪種方式進(jìn)行消息發(fā)送，需要根據(jù)情況進(jìn)行判斷，下面具體介紹不同的發(fā)送方式實(shí)現(xiàn)的消息可靠性保證。
1 同步發(fā)送
同步發(fā)送是指發(fā)送端在發(fā)送消息時(shí)，阻塞線程進(jìn)行等待，直到服務(wù)器返回發(fā)送的結(jié)果。發(fā)送端如果需要保證消息的可靠性，防止消息發(fā)送失敗，可以采用同步阻塞式的發(fā)送，然后同步檢查Brocker返回的狀態(tài)來判斷消息是否持久化成功。如果發(fā)送超時(shí)或者失敗，則會(huì)默認(rèn)重試2次，RocketMQ選擇至少傳輸成功一次的消息模型，但是有可能發(fā)生重復(fù)投遞，因?yàn)榫W(wǎng)絡(luò)傳輸是不可靠的，具體的重試策略可以參照第四小節(jié)。
2 異步發(fā)送
異步發(fā)送是指發(fā)送端在發(fā)送消息時(shí)，傳入回調(diào)接口實(shí)現(xiàn)類，調(diào)用該發(fā)送接口后不會(huì)阻塞，發(fā)送方法會(huì)立即返回，回調(diào)任務(wù)會(huì)在另一個(gè)線程中執(zhí)行，消息發(fā)送結(jié)果會(huì)回傳給相應(yīng)的回調(diào)函數(shù)。具體的業(yè)務(wù)實(shí)現(xiàn)可以根據(jù)發(fā)送的結(jié)果信息來判斷是否需要重試來保證消息的可靠性。
3 單向發(fā)送
單向發(fā)送是指發(fā)送端發(fā)送完成之后，調(diào)用該發(fā)送接口后立刻返回，并不返回發(fā)送的結(jié)果，業(yè)務(wù)方無(wú)法根據(jù)發(fā)送的狀態(tài)來判斷消息是否發(fā)送成功，單向發(fā)送相對(duì)前兩種發(fā)送方式來說是一種不可靠的消息發(fā)送方式，因此要保證消息發(fā)送的可靠性，不推薦采用這種方式來發(fā)送消息。
4 發(fā)送重試策略
RocketMQ架構(gòu)模型中會(huì)有多個(gè)Borker為某個(gè)topic提供服務(wù)，一個(gè)topic下的消息分散存儲(chǔ)在多個(gè)Broker存儲(chǔ)端，它們是多對(duì)多關(guān)系。Broker會(huì)將其提供存儲(chǔ)服務(wù)的topic的元數(shù)據(jù)信息上報(bào)到NameServer，對(duì)等NameServer節(jié)點(diǎn)組成的高可用服務(wù)會(huì)維護(hù)topic與Broker之間的映射關(guān)系，多對(duì)多的映射關(guān)系為消息可以重試發(fā)送到多個(gè)Broker端提供了前提與基礎(chǔ)。
當(dāng)發(fā)送端需要發(fā)送消息時(shí)，如果發(fā)送端中緩存了topic的路由信息，并包含了消息隊(duì)列,則直接返回該路由信息,如果沒有緩存或沒有消息隊(duì)列，則向NameServer查詢?cè)搕opic的路由信息，查詢到路由消息之后，采用指定的隊(duì)列選擇策略選擇相應(yīng)的queue發(fā)送消息，默認(rèn)是采用輪詢策略，發(fā)送成功則返回, 收到異常則根據(jù)相應(yīng)的策略進(jìn)行重試，可以根據(jù)發(fā)送端感知到的Broker的時(shí)延、上次發(fā)送失敗的Broker信息和發(fā)送端配置的是否重試不同Broker的參數(shù)以及發(fā)送端設(shè)置的最大超時(shí)時(shí)間等等策略來靈活地實(shí)現(xiàn)不同等級(jí)的消息發(fā)送可靠性保證。重試策略可以有效的保證消息發(fā)送成功的概率，最終提高消息發(fā)送的可靠性。
二 存儲(chǔ)端消息可靠性
RocketMQ的消息存儲(chǔ)結(jié)構(gòu)如下圖所示：
- 消息隊(duì)列存儲(chǔ)的最小單位是消息Message。
- 同一個(gè)Topic下的消息映射成多個(gè)邏輯隊(duì)列。
- 不同Topic的消息按照到達(dá)broker的先后順序以Append的方式添加至CommitLog，順序?qū)?，隨機(jī)讀。
目前RocketMQ存儲(chǔ)模型使用本地磁盤進(jìn)行存儲(chǔ)，數(shù)據(jù)寫入為producer -> direct memory -> pagecache -> 磁盤，數(shù)據(jù)讀取如果pagecache有數(shù)據(jù)則直接從pagecache讀，否則需要先從磁盤加載到pagecache中。Broker存儲(chǔ)節(jié)點(diǎn)的文件存儲(chǔ)模式如下圖所示：
Broker端CommitLog采用順序?qū)?，可以大大提高寫入效率，同時(shí)采用不同的刷盤模式提供不同的數(shù)據(jù)可靠性保證，此外采用了ConsumeQueue中間結(jié)構(gòu)來存儲(chǔ)偏移量信息，實(shí)現(xiàn)消息的分發(fā)。由于ConsumeQueue結(jié)構(gòu)固定且大小有限，在實(shí)際情況中，大部分的ConsumeQueue 能夠被全部讀入內(nèi)存，可以達(dá)到內(nèi)存讀取的速度。此外為了保證CommitLog和ConsumeQueue的一致性， CommitLog里存儲(chǔ)了Consume Queues 、Message Key、Tag等所有信息，即使ConsumeQueue丟失，也可以通過 commitLog完全恢復(fù)出來，這樣只要保證commitLog數(shù)據(jù)的可靠性，就可以保證Consume Queue的可靠性。
RocketMQ存儲(chǔ)端采用本地磁盤進(jìn)行CommitLog消息數(shù)據(jù)的存儲(chǔ)，不可避免的就會(huì)帶來存儲(chǔ)可靠性的挑戰(zhàn)，如何保證消息不丟失，RocketMQ消息服務(wù)一直在不斷提高數(shù)據(jù)的可靠性。
1 存儲(chǔ)可靠性挑戰(zhàn)
RocketMQ存儲(chǔ)端也即Broker端在存儲(chǔ)消息的時(shí)候會(huì)面臨以下的存儲(chǔ)可靠性挑戰(zhàn)：
- Broker正常關(guān)閉
- Broker異常Crash
- OS Crash
- 機(jī)器掉電，但是能立即恢復(fù)供電情況
- 機(jī)器無(wú)法開機(jī)(可能是cpu、主板、內(nèi)存等關(guān)鍵設(shè)備損壞)
- 磁盤設(shè)備損壞
1正常關(guān)閉，Broker 可以正常啟動(dòng)并恢復(fù)所有數(shù)據(jù)。2、3、4同步刷盤可以保證數(shù)據(jù)不丟失，異步刷盤可能導(dǎo)致少量數(shù)據(jù)丟失。5、6屬于單點(diǎn)故障，且無(wú)法恢復(fù)。解決單點(diǎn)故障可以采用增加Slave節(jié)點(diǎn)，主從異步復(fù)制仍然可能有極少量數(shù)據(jù)丟失，同步復(fù)制可以完全避免單點(diǎn)問題。
這里一般來說就需要在性能和可靠性之間做出取舍，對(duì)于RocketMQ來說，Broker的可靠性主要由兩個(gè)方面保障：
- 單機(jī)的刷盤機(jī)制
- 主從之間的數(shù)據(jù)復(fù)制
如果設(shè)置為每條消息都強(qiáng)制刷盤、主從復(fù)制，那么性能無(wú)疑會(huì)降低;如果不這樣設(shè)置，就會(huì)有一定的可能性丟失消息。RocketMQ一般都是先把消息寫到PageCache中，然后再持久化到磁盤上，數(shù)據(jù)從pagecache刷新到磁盤有兩種方式，同步和異步。整體的消息寫入和讀取如下圖所示：
針對(duì)broker端單機(jī)存儲(chǔ)可靠性，主要依賴單機(jī)的刷盤策略，主從之間的副本復(fù)制可以參考下一章節(jié)的主從模式。
2 同步刷盤
消息寫入內(nèi)存的 PageCache后，立刻通知刷盤線程刷盤，然后等待刷盤完成，刷盤線程執(zhí)行完成后喚醒等待的線程，返回消息寫成功的狀態(tài)。這種方式可以保證數(shù)據(jù)絕對(duì)安全，但是吞吐量不大。
3 異步刷盤(默認(rèn))
消息寫入到內(nèi)存的 PageCache中，就立刻給客戶端返回寫操作成功，當(dāng) PageCache中的消息積累到一定的量時(shí)，觸發(fā)一次寫操作，或者定時(shí)等策略將 PageCache中的消息寫入到磁盤中。這種方式吞吐量大，性能高，但是 PageCache中的數(shù)據(jù)可能丟失，不能保證數(shù)據(jù)絕對(duì)的安全。
實(shí)際應(yīng)用中要結(jié)合業(yè)務(wù)場(chǎng)景，合理設(shè)置刷盤方式，尤其是同步刷盤的方式，由于頻繁的觸發(fā)磁盤寫動(dòng)作，會(huì)明顯降低性能。
4 過期文件刪除
由于RocketMQ操作CommitLog、ConsumeQueue文件是基于文件內(nèi)存映射機(jī)制，并且在啟動(dòng)的時(shí)候會(huì)將所有的文件加載，為了避免內(nèi)存與磁盤的浪費(fèi)、能夠讓磁盤能夠循環(huán)利用、避免因?yàn)榇疟P不足導(dǎo)致消息無(wú)法寫入等引入了文件過期刪除機(jī)制。最終使得磁盤水位保持在一定水平，最終保證新寫入消息的可靠存儲(chǔ)。
三 消費(fèi)端消息可靠性
RockerMQ默認(rèn)提供了至少消費(fèi)一次的消費(fèi)語(yǔ)義來保證消息的可靠消費(fèi)。
通常消費(fèi)消息的確認(rèn)機(jī)制一般分為兩種思路：
- 先提交后消費(fèi)
- 先消費(fèi)，消費(fèi)成功后再提交
思路1可以解決重復(fù)消費(fèi)的問題但是會(huì)丟失消息，因此RocketMQ默認(rèn)實(shí)現(xiàn)的是思路2，由各自consumer業(yè)務(wù)方保證冪等來解決重復(fù)消費(fèi)問題。
消費(fèi)端Consumer消費(fèi)消息核心邏輯如下圖所示：
1 消費(fèi)重試
消費(fèi)者從RocketMQ拉取到消息之后，需要返回消費(fèi)成功來表示業(yè)務(wù)方正常消費(fèi)完成。因此只有返回CONSUME_SUCCESS才算消費(fèi)完成，如果返回CONSUME_LATER則會(huì)按照不同的messageDelayLevel時(shí)間進(jìn)行再次消費(fèi)，時(shí)間分級(jí)從秒到小時(shí)，最長(zhǎng)時(shí)間為2個(gè)小時(shí)后再次進(jìn)行消費(fèi)重試，如果消費(fèi)滿16次之后還是未能消費(fèi)成功，則不再重試，會(huì)將消息發(fā)送到死信隊(duì)列，從而保證消息存儲(chǔ)的可靠性。
2 死信隊(duì)列
未能成功消費(fèi)的消息，消息隊(duì)列并不會(huì)立刻將消息丟棄，而是將消息發(fā)送到死信隊(duì)列，其名稱是在原隊(duì)列名稱前加%DLQ%，如果消息最終進(jìn)入了死信隊(duì)列，則可以通過RocketMQ提供的相關(guān)接口從死信隊(duì)列獲取到相應(yīng)的消息，保證了消息消費(fèi)的可靠性。
3 消息回溯
回溯消費(fèi)是指Consumer已經(jīng)消費(fèi)成功的消息，或者之前消費(fèi)業(yè)務(wù)邏輯有問題，現(xiàn)在需要重新消費(fèi)。要支持此功能，則Broker存儲(chǔ)端在向Consumer消費(fèi)端投遞成功消息后，消息仍然需要保留。重新消費(fèi)一般是按照時(shí)間維度，例如由于Consumer系統(tǒng)故障，恢復(fù)后需要重新消費(fèi)1小時(shí)前的數(shù)據(jù)。RocketMQ Broker提供了一種機(jī)制，可以按照時(shí)間維度來回退消費(fèi)進(jìn)度，這樣就可以保證只要發(fā)送成功的消息，只要消息沒有過期，消息始終是可以消費(fèi)到的。
四 總結(jié)
本文從消息流轉(zhuǎn)的整個(gè)過程分析了RocketMQ如何保證消息的可靠性，消息發(fā)送通過不同的重試策略保證了消息的可靠發(fā)送，消息存儲(chǔ)通過不同的刷盤機(jī)制以及多副本來保證消息的可靠存儲(chǔ)，消息消費(fèi)通過至少消費(fèi)成功一次以及消費(fèi)重試機(jī)制來保證消息的可靠消費(fèi)，RocketMQ在保證消息的可靠性上做到了全鏈路閉環(huán)，最大限度的保證了消息不丟失。
