如何設計一個簡單的消息中間件

2020 年 9 月 5 日
筆記

前言

我們日常開發當中需要用到消息中間件的場合很多，我們或許也用到了形形色色的消息中間件產品，有老牌的ActiveMQ、RabbitMQ，炙手可熱的Kafaka，還有阿里研發的Notify、MetaQ、RocketMQ等等，但反過來思考一下，如果讓我們自己來設計一個消息中間件，需要考慮哪些方面的問題，需要有什麼樣的特性來滿足實際業務生產的需要呢？下面就這個問題展開討論。

消息隊列應該有什麼樣的特性

很多有經驗的工程師看到這個問題，腦袋裡最直接能想到的應該是：解耦、非同步、削峰。

解耦

想像這樣一種場景，有個業務系統A，在處理某個核心業務邏輯的時候，跟另外的B、C、D三個業務系統有關聯，當前已有的處理流程是A系統在處理完自己本地的事務後，順次調用B、C、D三個系統的介面去同步數據狀態。但是，隨著業務的發展，突然有一天另外一個業務系統D的數據也同樣需要根據A系統當中數據記錄的更新而同步更新，此時，開發不得不去修改原有程式碼，加上調用D系統同步更新數據的程式碼。然後某一天，業務需求又發生了變更，B系統不再維護對應的業務數據，開發又不得不修改原有程式碼，將調用B系統介面的程式碼給刪掉。如此周而復始，開發就會很苦惱。其實，站在A系統的角度來看，可能「關心」A系統變更的應用有很多個，但A系統只需要發布變更消息即可，誰關心誰接入。

非同步

另外一種情況就是，還是上面最初的場景，A系統處理業務邏輯時，需要調用B、C、D三個系統的介面，但是其中D介面是個耗時任務介面，需要很長的時間才會得到處理結果，那麼一旦這樣的請求多了以後，A系統和D系統都會被拖垮。一個典型的例子是，電商當中的訂單系統，訂單最終支付成功之後可能需要給用戶發送簡訊積分什麼的，但其實這已經不是我們系統的核心流程了。如果外部系統速度偏慢（比如簡訊網關速度不好），那麼主流程的時間會加長很多，用戶也肯定不希望點擊支付過好幾分鐘才看到結果。那麼我們只需要通知簡訊系統「我們支付成功了」，不一定非要等待它處理完成。

削峰

還是A系統，A系統裡面有個秒殺業務邏輯，每天上午11點有一波搶購活動，對於一天的其他時間來說，訪問A系統的請求流量平平淡淡，A系統完全可以應付的了，但就是11點時候，有一波「突襲」流量訪問進來，上游入口還好，做了集群部署等一系列應對高並發的措施，但MySQL資料庫扛不住，每秒2K個請求已經是極限了。

像上面這種情況，上下游處理能力存在明顯差距，利用消息隊列來做一個通用的「漏斗」，當下游有能力處理的時候，再進行分發，就是一種很好的處理方式。

實際上，除了上面三個典型的應用場景以外，消息隊列還有一個應用場景，那就是—最終一致性。

最終一致性

以一個銀行的轉賬過程來理解最終一致性，轉賬的需求很簡單，如果A系統扣錢成功，則B系統加錢一定成功。反之則一起回滾，像什麼都沒發生一樣。然而，這個過程中存在很多可能的意外：

A扣錢成功，調用B加錢介面失敗。
A扣錢成功，調用B加錢介面雖然成功，但獲取最終結果時網路異常引起超時。
A扣錢成功，B加錢失敗，A想回滾扣的錢，但A機器down機。

那麼，要解決上面提到的可能出現的意外，就有兩種備選方案：

用分散式事務去實現強一致性，但實際上這種實現成本很高。
利用消息隊列的「記錄」和「補償」的方式去實現最終一致性。

這裡主要講第二種實現方式。回到剛才的例子，系統在A扣錢成功的情況下，把要給B「通知」這件事記錄在庫里（為了保證最高的可靠性可以把通知B系統加錢和扣錢成功這兩件事維護在一個本地事務里），通知成功則刪除這條記錄，通知失敗或不確定則依靠定時任務補償性地通知我們，直到我們把狀態更新成正確的為止。

整個這個模型依然可以基於RPC來做，但可以抽象成一個統一的模型，基於消息隊列來做一個「企業匯流排」。具體來說，本地事務維護業務變化和通知消息，一起落地（失敗則一起回滾），然後RPC到達broker，在broker成功落地後，RPC返回成功，本地消息可以刪除。否則本地消息一直靠定時任務輪詢不斷重發，這樣就保證了消息可靠落地broker。 broker往consumer發送消息的過程類似，一直發送消息，直到consumer發送消費成功確認。我們先不理會重複消息的問題，通過兩次消息落地加補償，下游是一定可以收到消息的。然後依賴狀態機版本號等方式做判重，更新自己的業務，就實現了最終一致性。

最終一致性不是消息隊列的必備特性，但某些時候確實可以依賴消息隊列做一些需要滿足最終一致性的事情。那麼可以再思考一下，理論上只要消息隊列不能100%保證不丟消息，那也無法實現最終一致性。

如何設計一個消息隊列

其實總體而言，我們設計一個消息隊列，一言以蔽之，可以簡單的理解為設計一個整體的消息被消費的數據流。

其中主要涉及到三個角色：消息生產Producer、Broker(消息服務端)、消息消費者Consumer。

Producer(消息生產者)：發送消息到Broker。
Broker(服務端)：Broker這個概念主要來自於Apache的ActiveMQ，特指消息隊列的服務端。
Consumer(消息消費者)：從消息隊列接收消息，consumer回復消費確認。

其中，broker是我們的設計重點，它主要有三個職能：

消息的轉儲：消息存儲在broker伺服器上，在合適的時間點把消息投遞出去，或者通過一系列手段輔助消息最終能送達消費機。
規範一種範式和通用的模式，以滿足解耦、最終一致性、錯峰等需求。
消息的傳輸：RPC調用

所以，一個消息隊列的基本實現可以概括為：

build一個整體的數據流,例如producer發送給broker,broker發送給consumer,consumer回復消費確認，broker刪除/備份消息等。利用RPC將數據流串起來。然後考慮RPC的高可用性，盡量做到無狀態，方便水平擴展。之後考慮如何承載消息堆積，然後在合適的時機投遞消息，而處理堆積的最佳方式，就是存儲，存儲的選型需要綜合考慮性能/可靠性和開發維護成本等諸多因素。為了實現解耦非同步功能，我們必須要維護消費關係，可以利用zk/config server等保存消費關係。

當然，在基本實現的基礎之上，消息隊列也會視實際情況封裝一些高級特性，如可靠投遞，事務特性，性能優化等，這些高級特性不是本文探討的重點，本文主要關注消息隊列基本特性的原理和設計，即通訊協議、存儲選擇和消費關係維護這幾方面。

通訊協議

消息Message既是資訊的載體，消息發送者需要知道如何構造消息，消息接收者需要知道如何解析消息，它們需要按照一種統一的格式描述消息，這種統一的格式稱之為消息協議。

幾種常見消息通訊協議

JMS：JMS是由Sun公司早期提出的消息標準，旨在為java應用提供統一的消息操作，包括創建消息、發送消息、接收消息等。JMS提供了兩種消息模型，點對點和發布訂閱模型，當採用點對點模型時，消息將發送到一個隊列，該隊列的消息只能被一個消費者消費。而採用發布訂閱模型時，消息可以被多個消費者消費。在發布訂閱模型中，生產者和消費者完全獨立，不需要感知對方的存在。
AMQP：AMQP是 Advanced Message Queuing Protocol，即高級消息隊列協議。AMQP不是一個具體的消息隊列實現，而是一個標準化的消息中間件協議。其目標是讓不同語言，不同系統的應用互相通訊，並提供一個簡單統一的模型和編程介面。目前主流的ActiveMQ和RabbitMQ都支援AMQP協議。AMQP不從API層進行限定，而是直接定義網路交換的數據格式。
Kafka的通訊協議：Kafka的Producer、Broker和Consumer之間採用的是一套自行設計的基於TCP層的協議。Kafka的這套協議完全是為了Kafka自身的業務需求而訂製的。

存儲選型

通常來說，可供選擇的存儲類型有如下幾種：

記憶體
本地文件系統
分散式文件系統
DB
NoSQL

從速度上記憶體顯然是最快的，對於允許消息丟失，消息堆積能力要求不高的場景(例如日誌)，記憶體會是比較好的選擇。

DB則是最簡單的實現可靠存儲的方案，很適合用在可靠性要求很高，最終一致性的場景(例如交易消息)，對於不需要100%保證數據完整性的場景，要求性能和消息堆積的場景，hbase也是一個很好的選擇。

具體的選擇還是要從支援的業務場景出發作出最合理的選擇，如果你們的消息隊列是用來支援支付/交易等對可靠性要求非常高，但對性能和量的要求沒有這麼高，而且沒有時間精力專門做文件存儲系統的研究，DB是最好的選擇；對於不需要100%保證數據完整性的場景，要求性能和消息堆積的場景，hbase也是一個很好的選擇，典型的比如 kafka的消息落地可以使用hadoop。

消費關係處理

經過上面的存儲選型以後，我們的消息隊列就初步具備了轉儲消息的能力。下面一個重要的事情就是解析發送接收關係，進行正確的消息投遞了。市面上的消息隊列定義了一堆讓人暈頭轉向的名詞，如JMS 規範中的Topic/Queue，Kafka裡面的Topic/Partition/ConsumerGroup，RabbitMQ裡面的Exchange等等。掰開了揉碎了看，無外乎是單播與廣播的區別。所謂單播，就是點到點；而廣播，是一點對多點。

為了實現廣播功能，我們必須要維護消費關係，通常消息隊列本身不維護消費訂閱關係，可以利用zookeeper等成熟的系統維護消費關係，在消費關係發生變化時下發通知。

最後

以上就是一個基本的消息隊列設計需要考慮的特性和關鍵點，大家get了多少呢，歡迎留言一起探討。歡迎關注我：野生技術匯