《數據密集型應用系統設計》消息代理

引言

消息代理其實指的就是消息隊列，但是我認為作者這裏的代理，是給予系統架構位置考量的，因為消息中間件的本質就是作為不同服務之間交流的一種媒介。

介紹

消息代理可以看作是 處理數據流進行優化的數據庫。

消息代理通常部署在獨立的服務器當中，無論是生產者還是消費者，都有可能來自於不同的服務。整個流程通常為生產者生產數據通過消息代理當中，消費者連接消息代理接受生產者數據進行消費。

消息存在在中間代理有一個明顯的好處是可以屏蔽頻繁變動的生產者端和消費者端，將有關消息代理內部的特性轉移到代理中。

比如是否持久化問題，有的消息代理處理消息方式是無論是否消費都會存盤（Rocket MQ），保證消息不會隨服務器的宕機而丟失消息數據。

當然也有比較粗暴的消息代理處理方式，把消息放在內存中，一旦關閉立馬釋放消息，但是這樣也會導致消息丟失。

消息代理的優勢和劣勢都在異步處理，生產者只需要確保生產數據正確發送並且正確存儲到消息代理中，這些步驟處理完成之後，生產者可以接着處理其他業務。

而消費者則不同，消費者可以配置定期獲取消息代理消息並且檢查消息內容是否屬於自己消費範疇，這保證了消費者可以及時處理，但是處理消費時效性是不確定的（幾分鐘、幾小時、甚至幾天），如果消費者的消費能力或者處理效率過低，就會出現消息擠積壓的問題。

大部分情況下可以使用無限隊列積壓消息和切換消費的者的方式對待消費慢的消費者。

消息代理對比數據庫

現在一部分消息代理設計可以使用兩階段提交，看起來似乎越來越和數據庫進行靠攏。

消息代理雖然可以看作是優化數據流的數據庫，但是消息隊列和數據庫是存在本質差距的，主要的差距如下：

• 數據庫需要保證數據的持久化，刪除需要指定的命令完成，否則不能擅自丟失數據。傳統的消息代理更多設計為消息成功傳遞立馬刪除消息，這樣被消費過的消息就不會堆積，也不會有重複消費問題。（但是有部分消息代理存在特例）
• 消息代理刪除消息，多數消息的工作區間非常消息，隊列也比較短，通常這些內容都能很快的在內存中"轉接"，但是一旦消息堆積，消息無法在內存中堆放，就需要臨時序列化持久存儲到磁盤中，等待內存有了足夠空間之後再加載內容。這一步操作需要消耗系統的CPU和IO資源。
• 數據庫通常會使用多級索引加快數據的搜索，而消息代理通常支持某種消息模型來支持特定主題主題發送模式，以及利用日誌順序讀寫加快數據搜索。
• 數據庫查詢數據通常基於數據的時間點快照，為了保證數據的ACID特性，數據庫通常需要保證前後兩個線程之間的數據可見性正常，前者在不重複查詢的前提下，不應該看到後者改動的數據內容（當然也可以做到完全看不到，比如單線程化）。消息代理則側重於在數據改動之後通知客户端，對於消息查詢的能力支持較弱（或乾脆沒有）。

多個消費者讀取

生產者端的數據處理 通常比較簡單，消息代理的關注重點再對待消費者的“消費行為”上， 目前消息代理有兩種主要的消費模式：負載均衡和扇出式。

負載均衡

負載均衡代表了每一個消息只能傳給一個消費者，消費者可以共享處理消息的動作，代理也可以分配給任意的消費者，消息處理的代價非常高的時候，這種負載均衡的模式比較受歡迎，通常我們會希望添加消費者並行處理消息。

扇出式

扇出式指的是消息會發給所有的消費者，使用的實現方式是讓獨立的消費者共同“監聽”相通的消費信息同時不互相進行干涉，也可以看作是一個流被複制到不同的批次裏面進行工作（實現方式通過JMS和AMQP進行交換綁定）。

扇出式和負載均衡方式可以組合完成，比如可以通過負載均衡的多個分組而消費組內每一個消費者都可以接受消息。

消息傳遞確認

消息傳遞過程具有不確定性，消費者接受到消息有可能出現不會處理，或者無法處理的情況。為了確保消息不會丟失，客户端再進行消息處理之後必須告訴消息代理，然後消息代理才能確認是否真的被消費過。

如果連接超時或者客户端處理超時，消息代理沒有收到消息代理處理完成的請求，則需要把消息重新傳遞給另一個消費者，通常會把消息重傳多次，直到成功為止，否則就認為是存在消費失敗的情況。

還有一種情況是有的時候可能接收到消息的客户端已經把消息處理過了，但是在通知消息代理之前崩潰了，這時候就涉及分佈式事務問題。

重複消費

消息傳遞的最後一個問題是重複消費問題：

以上面的圖為例，如消費者2在消費m3的時候突然發生崩潰，此時消費者1,剛好消費m4消費完畢，然而下一個消費確實意料之外的消息m3，最後才是消費m5。出現這樣的情況由於消費者2消費m3的時候沒有對消息代理進行迴應。

解決負載均衡和重複消費問題，通常的解決方案是使用單獨隊列的方式處理，但是如果消息和消息之間沒有交集，則完全可以放心重排序問題。

分區日誌

傳統的消息隊列通常不具備和數據庫相同的功能，消費者消費完消息之後，消息代理會把消息直接丟棄，早期消息代理是單純為了瞬時數據處理而出現的。

數據庫的設計思路則要求數據的持久化存儲，只要用户存在操作，應該是永久貯存到消息代理當中。

傳統消息隊列如果無法找到消費者，則通常會直接把消息丟失並且無法恢復，但是現代消息隨着量級膨脹會出現消息阻塞的問題。既然數據庫擅長存儲，而消息代理擅長數據的瞬時傳遞，那麼肯定是可以混合使用的。

日誌消息存儲後續受到消息隊列歡迎。

日誌存儲隊列

日誌存儲結構的關鍵是 順序讀寫 和 追加，在[[《數據密集型型系統設計》LSM-Tree VS BTree]]中介紹了有關日誌存儲結構的特點。

日誌存儲和消息代理結合之後的工作方式是生產者推送到消息隊列使用追加日誌，而消費者則讀取最新日誌進行接收處理，如果讀取到末尾則立刻進入阻塞等待的狀態等待生產請求。

為了實現這樣的等待消費機制，消息代理通常會設計序號或者消費進度（偏移量）來完成每個消費者的消費進度監控。

在Unix系統當中tail -f 以相同的思路進行工作，默認情況下會監聽某個文件的末尾位置監聽改變。

使用序號遞增是由於日誌是隻追加不修改的，序號可以保證消息的發送和消費順序，但是如果使用分區併發發送，依然沒法保證順序消費。

下面的圖就是消息隊列消費的分區以及日誌存儲結合。

從上面的圖可以看到，生產者按照順序追加到分區後面，消費者維護偏移量記錄消費位置，但是分區的實際消費順序是無法保證的， 只能保證單個分區的消費順序 。

目前主流的消息隊列都會結合日誌消息存儲實現高可用，高可用是現代架構的基本要求，使用分區以及消息的順序讀寫可以基本達到媲美內存的操作速度，在實現百萬消息吞吐量的同時通過日誌存儲保持高可用以及保證消息的容錯性。

日誌和傳統消息對比

• 基於日誌的消息代理使用日誌的方式可以很好的支持扇出結構。
• 日誌消息代理支持負載均衡，可以把整個分區交給消費者進行消費，不需要將單個消費者給消費者客户端。

• 負載均衡的情況下每個客户端分配分區中的所有消息，之後將會通過單線程順序消費的方式對於分區進行消費處理。但是這樣會帶來一個問題，那就是一個主題最終只能有一個分區進行處理，並且只能有一個消費者進行消費，這樣會帶來兩個方面的問題。

  • 主題數量將會等於分區的數量，負載均衡變為單節點，消息隊列的所有優點被屏蔽了。
  • 如果消費者無法及時消費，將會出現消息堆積問題。

通過上面幾點我們可以瞭解到，針對不同的應用場景，選擇消息的處理方式也不一樣：

• 如果消息的處理代價非常高但消息的排序不是非常重要，可以並行處理保證消息的正常消費。這時候使用傳統的 JMS/AMQP 類型的消息代理進行處理。（舉例：日誌）
• 如果消費順序非常重要，消息順序也非常重要，使用日誌方式處理也可以很好工作。（舉例：扣款結算）

消費偏移量

順序讀取的方式可以容易實現偏移量處理的需求，通過加入消息偏移量，小於當前消費者偏移量的消息都可以認為已經被消費，而更大的偏移量此時消費者並沒有發現，在這種工作模式下，消費者只需要定期進行偏移量檢查，然後進行消費和移動偏移量即可。

這種方式有點類似流水線工作上下游定期查看上游推送的任務。

偏移量處理在主從複製的數據庫處理方式中的日誌序列號比較常見，比如Mysql 的Binlog日誌文件複製，主從複製支持從節點重新連接主節點之後，從斷開節點的日誌序列號開始重新進行同步，在不跳過任何寫入的情況下恢復節點複製。

如果消息消費失敗，通常由另一個節點負責接管工作，同時記錄偏移量最後的使用記錄。但是消費偏移量也會出現重複消費的情況，那就是消費者已經把消息消費處理完成了，但是移動偏移量的時候沒有記錄就崩潰了，那麼這條消息在消息隊列恢復之後會認為被消費！這種情況下會便出現了重複消費的情況。

磁盤空間使用

日誌消息存儲的關鍵問題是磁盤空間會隨着日誌記錄被耗盡，為了更好的管理消息日誌， 日誌設計通常會採用分區分段的方式，定期將舊段進行歸檔或者刪除。

日誌通常可以看作是非常大的緩衝區，緩存區一旦滿了通常需要刪除掉最早的數據，緩衝區通常會以環形緩衝或者被叫做循環緩衝的方式存在。如果消息的消費速度跟不上消息的生產速度，有可能出現消費者未進行消費，但是待刪除標記指向未消費片段的情況。

現代磁盤的存儲效率基本可以維持海量數據的日誌保存幾天甚至幾周的時間，不管消息的留存時間多長，可以確定的是消息一定會被存儲在磁盤形成的緩衝區當中，所以整個日誌的吞吐量基本恆定不變。

磁盤空間利用的另一種方式是當隊列過大的時候才把存在內存的消息寫入導磁盤當中，和固定放在磁盤緩衝區的日誌消息存儲相比，使用這種方式在內存的寫入是相當快的，但是磁盤讀寫的時候會明顯降速，這會導致隊列不能保證吞吐量的穩定（當然也取決於隊列中的消息數量）。

消費者跟不上生產者

上面基本討論了消費者跟不上生產者的的一系列問題，我們可以總結為三種處理方式：

• 丟棄消息
• 消息緩衝
• 應用抗壓

使用比較多的方式是日誌作為消息緩衝，因為緩衝要求具備較大並且固定大小的緩衝。（受到磁盤制約）

但是消費者落後太多，所需的信息如果比磁盤上的信息還要舊，那麼可能無法讀者這些信息。所以代理丟棄的消息是緩衝區容量不能容納的舊消息，實現的方式是監控消費者落後日誌頭部的距離，落後非常多的情況下需要進行報警，防止消息積壓。

消費者所消費的隊列“爆滿”之前，報警機制以及一些監控工具能提前判斷出消費者異常，緩衝區承載量通常足夠抗到解決問題修復完成，出現這些問題更多的情況是由於長事務或者長業務處理導致的阻塞問題，這些問題需要在消息消失之前及時處理修復完成。

那如果真的沒來得及處理並且開始丟失消息怎麼辦？ 通常也是消費者出現問題，消費者通常為集羣部署的（需要強一致性和順序消費例外），如果單消費者出現故障，可以通過異常之後立即下線的方式把消息轉移給其他消費者處理，同時現代消息隊列通常有心跳檢測和負載均衡，可以在某個節點故障的時候，自動切換到其他消費者進行消費，切換和異常處理有許多策略可以選擇。

傳統消息隊列則沒有這樣的好處，因為缺乏日誌管理，當消費者堆積消息，或者消費者節點宕機，消息依然會累計再消費者身上，待重啓之後依然需要處理分配給自己的任務，這顯然非常麻煩。

個人建議傳統消息隊列結合數據庫做一套符合業務的重試機制更為妥當。

重新處理消息

傳統消息隊列的重複消費是採用“即用即刪”的方式消費消息，如果消息消費失敗，通常的情況是消息被丟失刪除，所以傳統消息處理不會出現重複消費問題。

使用日誌的消息消費則安全很多，雖然會還是會出現重複消費問題，但是換來的是消息可以從日誌獲取到歷史記錄，同時因為是隻讀操作對於日誌也不會有任何干涉。

偏移量是消費者的巨大特權，消費可以通過移動偏移量的方式將過往消費信息重複消費，同時輸出到不同位置，手動進行數據的重複處理，可以類似批處理的方式，對於一個消息進行重複消費。

# 小結

在這本書中，作者將消息隊列統稱為消息代理，當然我們接觸更多的説法是消息中間件。

我們從消息代理和數據庫的區別開始，介紹了有關消息代理的巨大的優勢，瞬時處理和異步，它確保了兩個完全不同業務系統之間的消息安全傳遞，消息代理可以通過負載均衡或者扇出方式，將消息負載到多個消費者節點進行消費。

消息代理起初只是簡單的簡單的消息傳遞，並且消息不具備任何持久化存儲特點，在書中被稱之為AMQP/JMS的消息代理結構，這種結構中消息一旦接受成功處理會立馬刪除過期消息，保證內存有足夠的空間存放消息，不保證順序消費和消息響應而是側重於速度，這種方式更像是RPC的“升級”。

而現代的消息代理則多數使用日誌存儲結合，這種思路是從數據庫的持久化進行吸收融合，後續發展出基於日誌的消息存儲和消息回溯以及消費者高可用。消費者僅僅通過偏移量就可以檢查最新的消費內容，在宕機重啓之後從磁盤日誌中找到丟失消息進行重新接受和處理。

我們通常總是認為沾上日誌的日誌消息存儲要比傳統的AMQP消息隊裏的方式要慢，實際上這樣的想法是存在偏頗的，因為日誌採用的磁盤順序讀寫+追加寫入的方式，文件中的遊標不需要回溯，而磁盤順序讀寫+追加寫入基本可以保證效率基本可以和內存持平（至少不會斷崖式差距），所以日誌結構的存儲是現代消息代理的主流選擇。

消息隊列日誌的方式選擇和適合現代的互聯網移動生態十分契合，現代的網絡環境不再是單一化，而是隨着環境切換會切換不同的網絡，所以消息丟失和重複消費的事情幾乎是不可避免的（非人為干預的情況下）。

寫在最後

這一篇是抽取了整本書消息代理的部分，也就是設計了整個消息代理中比較核心的設計點和一些常見的設計產生的問題，只要入門任意一款現代主流的消息隊列中間件，基本能知道作者在表達的深層含義，所以建議看這部分內容之前，先了解一款消息中間件成品會有更好的閲讀體驗。