消息中間件應用的常見問題與方案
?1. 引言
消息隊列（MQ）中間件已經普及很多年了，在互聯網應用中，通常稍大一些的應用，我們都可以見到MQ的身影。當前市面上有很多中消息中間件，包括但不限于RabbitMQ、RocketMQ、ActiveMQ、Kafka(流處理中間件) 等。很多開發人員已經熟練地掌握了一個或者多個消息中間件的使用。但是仍然有一些小伙伴們對消息中間件不是特別熟悉，因為各種原因不能深入的去學習了解個中原理和細節，導致使用的時候可能出現這樣那樣的問題。在這里，我們就針對消息隊列中間件使用中的典型問題作一番分析（包括順序消息、可靠性保證、消息冪等、延時消息等），并提供一些解決方案。
2. 消息中間件應用背景
2.1 消息中間件基本思想
我們在單個系統中，一些業務處理可以順序依次的進行。而涉及到跨系統（有時候系統內部亦然）的時候，會產生比較復雜數據交互（也可以理解為消息傳遞）的需求，這些數據的交互傳遞方式，可以是同步也可以是異步的。在異步傳遞數據的情況下，往往需要一個載體，來臨時存儲與分發消息。在此基礎上，專門針對消息接收、存儲、轉發而設計與開發出來的專業應用程序，都可以理解為消息隊列中間件。
引申一下：如果我們自己簡單的使用一張數據庫表，來記錄數據，然后接受數據存儲在數據表，通過定時任務再將數據表的數據分發出去，那么我們已經實現了一個最簡單的消息系統（這就是本地消息表）。
我們可以認為消息中間件的基本思想就是 利用高效可靠的消息傳遞機制進行異步的數據傳輸。在這個基本思想的指導下，不同的消息中間，因為其側重場景目的不同，在功能、性能、整體設計理念上又各有差別。
消息隊列（MQ）本身是實現了生產者到消費者的單向通信模型，RabbitMQ、RocketMQ、Kafka這些常用的MQ都是指實現了這個模型的消息中間件。目前最常用的幾個消息中間件主要有，RabbitMQ、RocketMQ、Kafka（分布式流處理平臺）、Pulsar（分布式消息流平臺）。這里我將兩個流處理平臺納入其中了, 更早的一些其他消息中間件已經慢慢淡出視野。業務選型的時候我們遵循兩個主要的原則：最大熟悉程度原則（便于運維、使用可靠）、業務契合原則（中間件性能可以支撐業務體量、滿足業務功能需求）。
這幾個常用的消息中間件選型對比，很容易找到，這里就不詳細描述了。大概說一下：Pulsar目前用的不如 RabbitMQ、RocketMQ、Kafka多。RabbitMQ主要偏重是高可靠消息，RocketMQ性能和功能并重，Kafka主要是在大數據處理中應用比較多（Pulsar比較類似）。
2.2 引入消息中間件的意義
我們先簡單舉例介紹一下異步、解耦、削峰的意義與價值（參考下面這張流程圖）：
對于一個用戶注冊接口，假設有2個業務點，分別是注冊、發放新人福利，各需要50ms去處理邏輯。如果我們將這兩個業務流程耦合在一個接口，那么總計需要100ms處理完成。但是該流程中，用戶注冊時候，可以不用關心自己的福利是否立即發放，只要盡快注冊成功返回數據即可，后續新人福利這一部分業務可以在主流程之外處理。我們如果將其剝離出來，接口主流程中只處理登陸邏輯，并通過MQ推送一條消息，通過異步方式處理后續的發放新人福利邏輯，這樣即可保證注冊接口50ms左右即能獲取結果。
而發放新人福利的業務，則通過異步任務慢慢處理。通過拆分業務點，我們已經做到解耦，注冊的附屬業務中增加或減少功能點都不會影響主流程。另外如果一個業務主流程在某個點請求并發比較高，正好通過異步方式，可以將壓力分散到更長的時間段中去，達到減輕固定時間段處理壓力的目的，這就是流量削峰。
另外，單線程模型的語言，通常對消息中間件的需求更強烈。多線程模型的語言，或者協程型語言，雖然可以通過自身的多線程(或協程)機制，來實現業務內部的異步處理，但是考慮到持久化問題以及管理難度，還是成熟的中間件更適合用來做異步數據通信，中間件還能實現分布式系統之間的數據異步通信。
2.3 消息中間件的應用場景
消息中間件的應用場景主要有：
異步通信：可以用于業務系統內部的異步通信，也可以用于分布式系統信息交互。
系統解耦：將不同性質的業務進行隔離切分，提升性能，主附流程分層，按照重要性進行隔離，減少異常影響。
流量削峰：間歇性突刺流量分散處理，減少系統壓力，提升系統可用性。
分布式事務一致性：RocketMQ提供的事務消息功能可以處理分布式事務一致性（如電商訂單場景）。當然，也可以使用分布式事務中間件。
消息順序收發：這是最基礎的功能，先進先出，消息隊列必備。
延時消息：延遲觸發的業務場景，如下單后延遲取消未支付訂單等。
大數據處理：日志處理，kafka。
分布式緩存同步：消費MySQLbinlog日志進行緩存同步，或者業務變動直接推送到MQ消費。
所以，如果你的業務中有以上列舉的場景，或者類似的功能、性能需求，那么趕快引入「消息中間件」來提升你的業務性能吧。
3. 引入消息中間件帶來的一系列問題
雖然消息中間件引入有以上那么多好處，但是使用的時候依然會存在很多問題。例如：
- 引入消息中間件增加了系統復雜度，怎么使用維護；
- 消息發送失敗怎么辦（消息丟失）；
- 為了確保能發成功，消息重復發送了怎么辦（消息重復）；
- 消息在中間件流轉出現異常怎么處理；
- 消息消費時候，如果消費流程失敗了怎么處理，還能不能重新從中間件獲取到這條消息；
- 消費失敗如果還能獲取，那會不會出現失敗情況下，一直重復消費同一條消息，從而流程卡死；
- 消費失敗如果不能再獲取，那么我們該怎么確保這條消息能再次被處理；
- 重復消費到相同的消息流程怎么處理，會不會導致業務異常；
- 那么我們該怎么確保消費流程只成功執行一次；
- 對于那些有順序的消息我們應該怎么保證發送和消費的順序一致；
- 消息太多了，怎么保證消費腳本消費速度，以便更得上業務的處理需求，避免消息無限積壓；
- 我想要發送的消息，等上幾秒鐘的時間再消費到，該怎么做；
當然我們對于以上的這些問題，針對業務開發者來說，可以進行提煉，得到以下幾個重點問題：
- 消息順序性保證
- 避免消息丟失
- 消息的重復問題
- 消息積壓處理
- 延遲消息處理
4. 問題的解決方案
4.1 消息順序性保證
常規的消息中間件和流處理中間件，本身設計一般都能支持順序消息，但是根據中間件本身不同的設計目標，有不同的原理架構，導致我們業務中使用中間件的時候，要針對性做不同的處理。
以下幾個常用消息或流中間件的順序消息設計以及使用中亂序問題分析：
RabbitMQ:
RabbitMQ的單個隊列（queue）自身，可以保證消息的先進先出，在設計上，RabbitMQ所提供的單個隊列數據是存儲在單個broker節點上的，在開啟鏡像隊列的情況下，鏡像的隊列也只是作為消息副本而存在，服務依然由主隊列提供。這種情況下在單個隊列上進行消費，天然就是順序性的。不過由于單個隊列支持多消費者同時消費，我們在開啟多個消費者消費統一隊列上的數據時候，消息分散到多個消費者上，在并發高的時候，多個消費者無法保證處理消息的順序性。
解決方法就是對于需要強制順序的消息，使用同一個MQ隊列，并且針對單個隊列只開啟一個消費者消費（保證并發處理時候的順序性，多線程同理）。由此引發的單個隊列吞吐下降的問題，可以采取kafka的設計思想，針對單一任務開啟一組多個隊列，將需要順序的消息按照其固定標識（例如：ID）進行路由，分散到這一組隊列中，相同標識的消息進入到相同的隊列，單個隊列使用單個消費者消費，這樣即可以保證消息的順序與吞吐。
如圖所示：
Kafka：
Kafka是流處理中間件，在其設計中，沒有隊列的概念，消息的收發依賴于Topic，單個topic可以有多個partition（分區），這些partition可以分散到多臺broker節點上，并且partition還可以設置副本備份以保證其高可用。
Kafka同一個topic可以有多個消費者，甚至消費組。Kafka中消息消費一般使用消費組（消費組可以互不干涉的消費同一個topic下的消息）來進行消費，消費組中可以有多個消費者。同一個消費組消費單個topic下的多個partition時，將由kafka來調節消費組中消費者與partiton的消費進度與均衡。但是有一點是可以保證的：那就是單個partition在同一個消費組中只能被一個消費者消費。
以上的設計理念下，Kafka內部保證在同一個partition中的消息是順序的，不保證topic下的消息的順序性。Kafka的消息生產者發送消息的時候，是可以選擇將消息發送到哪個partition中的，我們只要將需要順序處理的消息，發送到topic下相同的partition，即可保證消息消費的順序性。（多線程語言使用單個消費者，多線程處理數據時，需要自己去保證處理的順序，這里略過）。
RocketMQ：
RocketMQ的一些基本概念和原理，可以通過阿里云的官網做一些了解： 什么是消息隊列RocketMQ版？- 消息隊列RocketMQ版 - 阿里云【1】 。
RocketMQ的消息收發也是基于Topic的，Topic下有多個 Queue， 分布在一個或多個 Broker 上，用來保證消息的高性能收發（ 與Kafka的Topic-Partition機制 有些類似，但內部實現原理并不相同 ）。
RocketMQ支持局部順序消息消費，也就是保證同一個消息隊列上的消息順序消費。不支持消息全局順序消費，如果要實現某一個主題的全局順序消息消費，可以將該主題的隊列數量設置為1，犧牲高可用性。具體圖解可以參考阿里云文檔: 順序消息2.0 - 消息隊列RocketMQ版 - 阿里云【2】
4.2 避免消息丟失
消息丟失需要分為三部分來看：消息生產者發送消息到消息中間件的過程不發生消息丟失，消息在消息中間件中從接受存儲到被消費的過程中消息不丟失， 消息消費的過程中保證能消費到中間件發送的消息而不會丟失。
生產者發送消息不丟失：
消息中間件一般都有消息發送確認機制（ACK）, 對于客戶端來說，只要配置好消息發送需要ACK確認，就可以根據返回的結果來判斷消息是否成功發送到中間件中。這一步通常與中間件的消息接受存儲流程設計有關系。根據中間件的設計，我們通常采取的措施如下：
- 開啟MQ的ACK（或confirm）機制，直接獲知消息發送結果
- 開啟消息隊列的持久化機制（落盤，如果需要特殊設置的話）
- 中間件本身做好高可用部署
- 消息發送失敗補償設計（重試等）
在具體的業務設計中，如果消息發送失敗，我們可以根據業務重要程度，做相應的補償，例如：
- 消息失敗重試機制（發送失敗，繼續重發，可以設置重試上限）
- 如果依然失敗，根據消息重要性，選擇降級方案：直接丟棄或者降級到其他中間件或載體（同時需要相應的降級補償推送或消費設計）
消息中間件消息不丟失：
數消息中間件的消息接收存儲機制各不相同，但是會根據其特性設計，最大限度保證消息不會丟失：
RabbitMQ消息接收與保存：
- RabbitMQ 消息發送可以開啟發送者confirm模式，所有消息是否發送成功都會通知發送者；
- 需要開啟隊列消息持久化保證消息落盤；
- RabbitMQ通過鏡像隊列來保證消息隊列的高可用，但是鏡像隊列只有Master提供服務，其他slave只提供備份服務；
- master宕機會從slave中選擇一個成為新的master提供服務；
- master的生產與消費的最新狀態都會廣播到slave；
RocketMQ消息接受與保存：
- RocketMQ普通消息發送有三種方式：同步（Sync）發送、異步（Async）發送和單向（Oneway）發送，其區別與準確性保證可以參看 「 發送普通消息（三種方式） - 消息隊列RocketMQ版 - 阿里云」 【3】
- 具體的RocketMQ內部設計的HA機制是主從同步機制，消息發送到Topic下并具體消息隊列的Master Broker中后，會將消息同步到Slave。
- 只有Master Broker才可以接收生產者發送的消息。而消費者，可以從Master也可以從Slave拉取并消費消息。
Kafka在消息接受到保存所做的設計有：
- 分區副本方式的設計保證消息的高可用，在創建topic的時候都可以設置分區副本的數量；
- 生產者可以選擇接收不同類型的確認（ACK），比如在消息被完全提交時候（寫入所有同步副本）的確認，或者在消息被寫入首領副本時的確認，或者在消息被發送到網絡時確認；
- Kafka的消息，寫入分區的時候僅僅是保存在某幾個分區副本文件系統內存中，并不是直接刷到磁盤了，因此宕機時候，單個副本仍然可能丟失數據。Kafka不能保證單個分區副本的數據一定不丟失，而是靠分區副本機制來確保消息的完善性（分布到不同的broker上）。
積壓消息保存時效問題
- Kafka對于topic下的數據，有容量上限、時間上限兩種消息存儲上限規則，觸發其中任何一個規則，都會刪除淘汰之前的消息。這個尤其需要注意。
- RocketMQ，消息在服務器存儲時間也有上限，達到上限的消息將會被刪除。也需要做相應的考量。
- 受持久化磁盤容量的影響，存儲積壓的數據不能超過磁盤的上限。
- 如果業務消費有異常，需要給足充足的冗余量，避免因為消費不及時而丟失數據。
消費者消費消息不丟失：
- 消息消費時候，也要開啟相應的ACK機制，消息消費成功即ACK（對于Kafka就是更新消費的offset）；
- 對于RocketMQ這種有消息重新消費設計的，需要設置最大消費次數，嘗試失敗的消息重復消費。
消息ACK帶來兩個問題：
- 消息消費失敗如果不能ACK可能會導致消息消費無限阻塞在某條消息處；
- 消息失敗重新消費導致消息消費重復。
無限阻塞的問題，可以參考RocketMQ消費失敗的重試機制，對消息重試做一定的設計：
- 在消息體上設計重試次數的屬性，消費失敗的消息增加重試次數后重新發送到中間件，等待下一次消費，本次消費成功發回消息直接ACK。
- 消息重試次數達到上限之后，如果仍不能成功，則啟用降級方案，將消息存儲到異常信息持久化載體如DB中。
- 手動或者定時任務補償處理失敗的消息。
消息重復消費問題參考下一個小節。
4.3 消息的重復問題（消費冪等）
在分析常用中間件的時候，我們往往會發現，中間件設計者將這個問題的處理，下放給中間件使用者，也就是業務開發者了。誠然，業務消費處理的邏輯比消息生產者復雜得多。生產者只需要保證將消息成功發送到中間件即可，而消費者需要在消費腳本中處理各種復雜的業務邏輯。
解決消息重復消費的問題，核心是使用唯一標識，來標記某條消息是否已經處理過。具體方案可選的則有很多，比如：
- 使用數據庫自增主鍵，或者唯一鍵來保證數據不會重復變動；
- 使用中間狀態，以及狀態變動有序性來判斷業務是否以已經被處理；
- 利用一張日志表來記錄已經處理成功的消息的 ID，如果新到的消息 ID 已經在日志表中，那么就不再處理這條消息；
- 或者消息唯一標識，在Redis等NoSQL中維護一個處理緩存，判斷是否已經處理過；
- 如果消費者業務流程比較長，則需要開發者自己保證整個業務消費邏輯中數據處理的事務性。
4.4 消息積壓處理
通常我們在引入消息中間件的時候，已經會評估與測試消息消費的生產與消費速率，盡量使其達到平衡。但業務也有一些不可預知的突發情況，可能會造成消息的大量積壓。在這個時候，我們可以采取如下的方式，來做處理：
臨時緊急擴容
- 通過增加消費腳本的方式，提升消費速率，如果下游沒有限制的話，可以很快的減少消息積壓。
- 如果消費者下游數據處理能力有限，我們可以考慮建立臨時隊列，通過臨時腳本，將消息快速轉移到臨時隊列，優先保證線上業務能順利貫通，而后開啟更多的消費腳本處理積壓的數據。（順序消息需要額外處理，并保證最終處理的順序）
- 優化消費腳本的處理速度，突破下游限制，如果有可能，可以考慮批量處理，下游擴容等方式。
消息積壓預防
- 做好業務設計與降級，避免產生無效消息占用資源
- 根據消息積壓程度，動態增減消費者數量，減少消息積壓
- 做好消息積壓處理緊急預案，異常情況根據預案設計，迅速針對處理
4.5 延遲消息處理
延遲消息這一項功能，在部分MQ中間件中有實現。延時消息和定時消息其實可以互相轉換。
RocketMQ：
RocketMQ定時消息不支持任意的時間精度(出于性能考量)。只支持特定級別的延遲消息。消息延遲級別在broker端通過messageDelayLevel配置。其內部對每一個延遲級別創建對應的消息消費隊列，然后創建對應延遲級別的定時任務，從消息消費隊列中將消息拉取并恢復消息的原主題和原消息消費隊列。
RabbitMQ:
RabbitMQ實現延遲消息通常有兩個方案：一是創建一個消息延遲死信隊列，搭配一個死信轉發隊列來實現消費延時。但是該方式如果前一個消息沒達到TTL時間，后一個消息即便達到了，也不會被轉發到轉發隊列中；另一個是使用延時Exchange插件（rabbitmq_delayed_message_exchange），消息在達到TTL之后才會轉發到對應的隊列中并被消費。
Kafka本身不支持延時消息或定時消息, 想要實現消息的延時，需要使用其他的方案。
借助數據庫與定時任務實現延時消息：
常用數據庫的索引結構都支持數據的順序索引。借助數據庫可以很方便地實現任意時間消息的延時消費。使用一張表存儲數據的消費時間，開啟定時任務，在滿足條件之后將該消息提取出來，后續轉發到順序隊列去處理或者直接處理都可以（已處理需要做標記，后續不再出現），但是直接處理需要考慮吞吐量和并發重復性等問題。不如單個腳本轉發到普通隊列去處理方便。數據庫支持的定時任務消息積壓是可控的，但是吞吐量會有局限。
借助Reids的有序列表實現延時消息：
Reids的有序列表zset結構，可以實現延時消息。將消息的消費時間作為分值，把消息添加到zset中。使用 zrangebyscore 命令消費消息 # 命令格式 zrangebysocre key min max withscores limit 0 1 消費最早的一條消息 #min max 分別表示開始的分值與結束的分值區間，分別使用 0和當前時間戳，可以查出達到消費時間的消息 # withscores 表示查詢的數據要帶分值。limit 后面 就是查詢的起始 offset 和數量 zrangebyscore key 0 {當前時間戳} withscores limit 0 1 。
當然，這個方案也有局限性，首先，redis必須配置持久化防止消息丟失（如果配置不合理不能100%保證，但是每個命令都持久化會造成性能下降，需要權衡）；其次，如果延時消息過多會造成消息的積壓形成大key；再次，需要自己做重復消費和消費失敗的平衡處理（當然有可能，還是建議開啟單個消費進程將延時消息轉移到普通隊列去消費）。
基于時間輪的任務調度：
在很多軟件中，都有基于時間輪實現定時任務的實現，使用時間輪以及多級時間輪可以實現延時任務調度。如果我們希望自己實現延時任務隊列，可以考慮使用此算法來實現任務的調度，但是需要自己根據具體的需求去設計支持任務的延時上限以及調度的時間粒度（多層級）。時間輪算法我這里就先不講解了，感興趣的可以自己去搜索了解。
5. 總結
通過以上幾個小節的介紹，相信各位已經能很自然的理解：消息隊列、異步解耦的功能與核心思想，并且對如何使用MQ來架構自己的業務有了一定的認知。大多數MQ使用中的問題，只是要求我們多思考，將細節思慮周到，以保證業務的高可用。甚至，我們還可以在這幾個解決方案中提煉一些核心出來，以便在業務中參照類似的思想，優化我們的業務。比如，消息順序性保證 其核心是順序消息生產者發送到唯一分區，再維持固定分區的單消費者順序消費；避免消息丟失的核心是每個步驟的確認與降級機制；消費冪等的核心是唯一性標識與步進狀態；消息積壓處理的核心是快速響應應急預案；延遲消息的核心是消息排序，優化點是性能提升。
科學的方法有歸納和演繹，學習問題處理方案的過程中，提煉出相應的核心思想，并在使用中演繹，將這些歸納總結的知識點，再應用到業務中去，更加得心應手的處理相應的事務，構建出高可用的業務架構，這才是我們最需要做到的。
