作者 | 修戟
來源 | 阿裡技術公衆号
分布式系統中一個重要的前提假設是所有的網絡傳輸都是不可靠的，在網絡傳輸不可靠的情況下，保證消息的可靠傳輸，除了進行重試投遞别無他法。常用的絕大多數消息隊列RocketMQ、RabbitMQ等在消息傳輸上都隻能保證至少傳輸成功一次，也即（At least once），而不能保證隻傳輸成功一次（Exactly once）。由于分布式系統網絡的不可靠，可能就會出現消息丢失的現象，那麼RocketMQ是如何最大限度的保證消息不丢失的呢？那就需要從消息的産生到最終消費的整個過程來分析，消息完整鍊路可以劃分為以下三個階段：
- 生産階段：消息在 Producer 發送端建立出來，經過網絡傳輸發送到 Broker 存儲端。
- 存儲階段：消息在 Broker 端存儲，如果是主備或者多副本，消息會在這個階段被複制到其他的節點或者副本上。
- 消費階段：Consumer 消費端從 Broker存儲端拉取消息，經過網絡傳輸發送到 Consumer 消費端上，并通過重試來最大限度的保證消息的消費。
一 發送端消息可靠性
發送端Producer發送消息Broker端的核心邏輯如下圖所示：
消息發送一般有以下幾種方式：同步發送、異步發送以及單向發送，業務具體選擇哪種方式進行消息發送，需要根據情況進行判斷，下面具體介紹不同的發送方式實作的消息可靠性保證。
1 同步發送
同步發送是指發送端在發送消息時，阻塞線程進行等待，直到伺服器傳回發送的結果。發送端如果需要保證消息的可靠性，防止消息發送失敗，可以采用同步阻塞式的發送，然後同步檢查Brocker傳回的狀态來判斷消息是否持久化成功。如果發送逾時或者失敗，則會預設重試2次，RocketMQ選擇至少傳輸成功一次的消息模型，但是有可能發生重複投遞，因為網絡傳輸是不可靠的，具體的重試政策可以參照第四小節。
2 異步發送
異步發送是指發送端在發送消息時，傳入回調接口實作類，調用該發送接口後不會阻塞，發送方法會立即傳回，回調任務會在另一個線程中執行，消息發送結果會回傳給相應的回調函數。具體的業務實作可以根據發送的結果資訊來判斷是否需要重試來保證消息的可靠性。
3 單向發送
單向發送是指發送端發送完成之後，調用該發送接口後立刻傳回，并不傳回發送的結果，業務方無法根據發送的狀态來判斷消息是否發送成功，單向發送相對前兩種發送方式來說是一種不可靠的消息發送方式，是以要保證消息發送的可靠性，不推薦采用這種方式來發送消息。
4 發送重試政策
RocketMQ架構模型中會有多個Borker為某個topic提供服務，一個topic下的消息分散存儲在多個Broker存儲端，它們是多對多關系。Broker會将其提供存儲服務的topic的中繼資料資訊上報到NameServer，對等NameServer節點組成的高可用服務會維護topic與Broker之間的映射關系，多對多的映射關系為消息可以重試發送到多個Broker端提供了前提與基礎。
當發送端需要發送消息時，如果發送端中緩存了topic的路由資訊，并包含了消息隊列,則直接傳回該路由資訊,如果沒有緩存或沒有消息隊列，則向NameServer查詢該topic的路由資訊，查詢到路由消息之後，采用指定的隊列選擇政策選擇相應的queue發送消息，預設是采用輪詢政策，發送成功則傳回, 收到異常則根據相應的政策進行重試，可以根據發送端感覺到的Broker的時延、上次發送失敗的Broker資訊和發送端配置的是否重試不同Broker的參數以及發送端設定的最大逾時時間等等政策來靈活地實作不同等級的消息發送可靠性保證。重試政策可以有效的保證消息發送成功的機率，最終提高消息發送的可靠性。
二 存儲端消息可靠性
RocketMQ的消息存儲結構如下圖所示：
- 消息隊列存儲的最小機關是消息Message。
- 同一個Topic下的消息映射成多個邏輯隊列。
- 不同Topic的消息按照到達broker的先後順序以Append的方式添加至CommitLog，順序寫，随機讀。
目前RocketMQ存儲模型使用本地磁盤進行存儲，資料寫入為producer -> direct memory -> pagecache -> 磁盤，資料讀取如果pagecache有資料則直接從pagecache讀，否則需要先從磁盤加載到pagecache中。Broker存儲節點的檔案存儲模式如下圖所示：
Broker端CommitLog采用順序寫，可以大大提高寫入效率，同時采用不同的刷盤模式提供不同的資料可靠性保證，此外采用了ConsumeQueue中間結構來存儲偏移量資訊，實作消息的分發。由于ConsumeQueue結構固定且大小有限，在實際情況中，大部分的ConsumeQueue 能夠被全部讀入記憶體，可以達到記憶體讀取的速度。此外為了保證CommitLog和ConsumeQueue的一緻性， CommitLog裡存儲了Consume Queues 、Message Key、Tag等所有資訊，即使ConsumeQueue丢失，也可以通過 commitLog完全恢複出來，這樣隻要保證commitLog資料的可靠性，就可以保證Consume Queue的可靠性。
RocketMQ存儲端采用本地磁盤進行CommitLog消息資料的存儲，不可避免的就會帶來存儲可靠性的挑戰，如何保證消息不丢失，RocketMQ消息服務一直在不斷提高資料的可靠性。
1 存儲可靠性挑戰
RocketMQ存儲端也即Broker端在存儲消息的時候會面臨以下的存儲可靠性挑戰：
- Broker正常關閉
- Broker異常Crash
- OS Crash
- 機器掉電，但是能立即恢複供電情況
- 機器無法開機（可能是cpu、主機闆、記憶體等關鍵裝置損壞）
- 磁盤裝置損壞
1正常關閉，Broker 可以正常啟動并恢複所有資料。2、3、4同步刷盤可以保證資料不丢失，異步刷盤可能導緻少量資料丢失。5、6屬于單點故障，且無法恢複。解決單點故障可以采用增加Slave節點，主從異步複制仍然可能有極少量資料丢失，同步複制可以完全避免單點問題。
這裡一般來說就需要在性能和可靠性之間做出取舍，對于RocketMQ來說，Broker的可靠性主要由兩個方面保障：
- 單機的刷盤機制
- 主從之間的資料複制
如果設定為每條消息都強制刷盤、主從複制，那麼性能無疑會降低；如果不這樣設定，就會有一定的可能性丢失消息。RocketMQ一般都是先把消息寫到PageCache中，然後再持久化到磁盤上，資料從pagecache重新整理到磁盤有兩種方式，同步和異步。整體的消息寫入和讀取如下圖所示：
針對broker端單機存儲可靠性，主要依賴單機的刷盤政策，主從之間的副本複制可以參考下一章節的主從模式。
2 同步刷盤
消息寫入記憶體的 PageCache後，立刻通知刷盤線程刷盤，然後等待刷盤完成，刷盤線程執行完成後喚醒等待的線程，傳回消息寫成功的狀态。這種方式可以保證資料絕對安全，但是吞吐量不大。
3 異步刷盤（預設）
消息寫入到記憶體的 PageCache中，就立刻給用戶端傳回寫操作成功，當 PageCache中的消息積累到一定的量時，觸發一次寫操作，或者定時等政策将 PageCache中的消息寫入到磁盤中。這種方式吞吐量大，性能高，但是 PageCache中的資料可能丢失，不能保證資料絕對的安全。
實際應用中要結合業務場景，合理設定刷盤方式，尤其是同步刷盤的方式，由于頻繁的觸發磁盤寫動作，會明顯降低性能。
4 過期檔案删除
由于RocketMQ操作CommitLog、ConsumeQueue檔案是基于檔案記憶體映射機制，并且在啟動的時候會将所有的檔案加載，為了避免記憶體與磁盤的浪費、能夠讓磁盤能夠循環利用、避免因為磁盤不足導緻消息無法寫入等引入了檔案過期删除機制。最終使得磁盤水位保持在一定水準，最終保證新寫入消息的可靠存儲。
三 消費端消息可靠性
RockerMQ預設提供了至少消費一次的消費語義來保證消息的可靠消費。
通常消費消息的确認機制一般分為兩種思路：
- 先送出後消費
- 先消費，消費成功後再送出
思路1可以解決重複消費的問題但是會丢失消息，是以RocketMQ預設實作的是思路2，由各自consumer業務方保證幂等來解決重複消費問題。
消費端Consumer消費消息核心邏輯如下圖所示：
1 消費重試
消費者從RocketMQ拉取到消息之後，需要傳回消費成功來表示業務方正常消費完成。是以隻有傳回CONSUME_SUCCESS才算消費完成，如果傳回CONSUME_LATER則會按照不同的messageDelayLevel時間進行再次消費，時間分級從秒到小時，最長時間為2個小時後再次進行消費重試，如果消費滿16次之後還是未能消費成功，則不再重試，會将消息發送到死信隊列，進而保證消息存儲的可靠性。
2 死信隊列
未能成功消費的消息，消息隊列并不會立刻将消息丢棄，而是将消息發送到死信隊列，其名稱是在原隊列名稱前加%DLQ%，如果消息最終進入了死信隊列，則可以通過RocketMQ提供的相關接口從死信隊列擷取到相應的消息，保證了消息消費的可靠性。
3 消息回溯
回溯消費是指Consumer已經消費成功的消息，或者之前消費業務邏輯有問題，現在需要重新消費。要支援此功能，則Broker存儲端在向Consumer消費端投遞成功消息後，消息仍然需要保留。重新消費一般是按照時間次元，例如由于Consumer系統故障，恢複後需要重新消費1小時前的資料。RocketMQ Broker提供了一種機制，可以按照時間次元來回退消費進度，這樣就可以保證隻要發送成功的消息，隻要消息沒有過期，消息始終是可以消費到的。
四 總結
本文從消息流轉的整個過程分析了RocketMQ如何保證消息的可靠性，消息發送通過不同的重試政策保證了消息的可靠發送，消息存儲通過不同的刷盤機制以及多副本來保證消息的可靠存儲，消息消費通過至少消費成功一次以及消費重試機制來保證消息的可靠消費，RocketMQ在保證消息的可靠性上做到了全鍊路閉環，最大限度的保證了消息不丢失。