揭祕RocketMQ新特性以及在金融場景下的實踐

做者：陳廣勝，Apache RocketMQ Committer，DeFiBus 創始人，微衆銀行技術專家，中間件平臺相關產品負責人，曾就任於 IBM 和華爲，負責運營商雲和大數據平臺建設。

原文：https://mp.weixin.qq.com/s/nfJXGZb402iQnXFcN6TZCg

前言

2019 年底， RocketMQ 正式發佈了 4.6.0 版本，增長了「 Request-Reply 」的同步調用的新特性。「 Request-Reply 」這個新特性是由微衆銀行的開發者們總結實踐經驗，並反饋給社區的。接下來本文會詳細介紹此新特性。

「 Request-Reply 」是什麼

圖1.1 「 Request-Reply 」模式

在以往的消息中間件的使用中， Producer 和 Consumer 只負責發送消息和消費消息，彼此之間不會通訊。而 「 Request-Reply 」模式容許 Producer 發出消息後，以同步或者異步的形式等待 Consumer 消費這條消息並返回一個響應消息，從而達到相似 RPC 的調用效果。在整個「 Request-Reply 」調用過程當中（簡稱RR調用）， Producer 首先發出一條消息，消息經由 Broker 被 Consumer 獲取並消費；Consumer 消費完這條消息後，會將針對該消息的響應做爲另一條消息發送出來，最終回到 Producer 。爲了便於描述，稱此時的 Producer 爲請求方，發出的消息爲「請求消息」；Consumer 稱爲服務方，返回的消息稱爲「響應消息」。

「 Request-Reply 」模式使得 RocketMQ 具有了同步調用的能力，拓展了 RocketMQ 的使用場景，使其具備更多的應用可能性。開發者能夠利用這個特性快速搭建本身的消息服務總線，實現 RPC 調用框架；因爲請求以消息的形式存儲在 Broker ，便於收集信息作調用鏈追蹤和分析；在微服務領域，也有着普遍的應用場景。

「 Request-Reply 」的實現邏輯

在 RR 調用中涉及到 Producer、Broker、Consumer 三個角色。

Producer 的實現邏輯

圖2.1 producer示意圖

一、對請求消息增長對應的標識

Producer 發送請求消息時，須要在消息的 Properties 裏增長RR調用的標識，其中關鍵的字段有 Correlation_Id、REPLY_TO_CLIENT。Correlation_Id 用來惟一標識一次RR請求，經過這個屬性來匹配同一個RR調用的請求消息和響應消息。REPLY_TO_CLIENT 用來標識請求消息的發出方，其值爲 Producer 的 ClientId 。

做爲請求方的 Producer 只需增長對應標識到消息中，在消息的發送邏輯上與原始 Producer 保持一致。

二、發完請求消息後等待響應消息。

請求方每次執行 Request 以後，會建立 RequestResponseFuture 對象，而且以 Correlation_Id 做爲key記錄到 ResponseFutureTable 中。執行 Request 的線程經過 RequestResponseFuture 裏定義的 CountDownLatch 實現阻塞。當響應消息回到 Producer 實例時，根據響應消息中的 Correlation_Id從ResponseFutureTable 中獲取對應地 RequestResponseFuture ，激活 CountDownLatch 喚醒阻塞的線程，執行對響應消息的處理。

圖2.2 RequestResponseFuture結構

Consumer 的實現邏輯

圖2.3 consumer示意圖

Consumer 只須要在正常消費一條請求消息後，建立響應消息併發送出去便可。建立響應消息時必須使用提供的工具類來建立，避免丟失 Correlation_Id、REPLY_TO_CLIENT 等標識和關聯RR請求的屬性。

Broker 的實現邏輯

圖2.4 Broker示意圖

Broker 對請求消息的處理與原先的處理邏輯同樣，可是對於響應消息則是採用主動 Push 的形式將消息推給請求方。服務方 Consumer 將響應消息發送到 Reply_topic 上， Broker 收到響應消息後會交由 ReplyMessageProcessor 處理。Processor 會將響應消息落到 CommitLog 中，而且根據響應消息中的 REPLY_TO_CLIENT 獲得請求方的 ClientId ，經過 ClientId 找到對應的 Producer 實例及其 Channel ，將響應消息直接推送給它。

全部的響應消息都會發送到 Reply_topic 上，這個 Topic 是由 Broker 自動建立的系統 Topic ，以「集羣名 _REPLY_TOPIC 」的格式命名。Reply topic 用於作路由發現，讓響應消息可以發回到請求消息來源的那個集羣，目的是保證響應消息回到的 Broker 是請求方有鏈接的 Broker 。採用 Broker 主動推送響應消息的目的也是爲了保證響應消息可以精準回到發出請求消息的實例上。

「 Request-Reply 」在金融場景下的實踐

金融業務要求服務要持續穩定，可以提供 7x24 小時穩定可用的服務，而且容錯能力要足夠強，對節點故障可以快速屏蔽影響，保證成功率，快速恢復。所以，微衆銀行根據具體的使用場景增長了應用多活、服務就近、熔斷等特性，構建了安全可靠的金融級消息總線 DeFiBus 。

圖3.1 總線架構圖

如圖所示， DeFiBus 自上而下分別是總線層、應用層、 DB 層。

總線層有兩個很是重要的服務，分別是 GNS 和 GSL 。對每一個客戶，會根據客戶信息而且按照權重分配到規劃好的 DCN 內，實現數據層面的分片。GNS 服務是在數據層面進行的分片尋址，肯定客戶所在的 DCN 。在服務層面，會將服務部署到不一樣的區域，在調用服務時會先訪問 GSL 服務，作服務層面的分片尋址，肯定當前要訪問的服務在哪一個 DCN 。從數據和服務兩個維度作分片，由 GNS 和 GSL 作分片尋址，最終由總線實現請求到 DCN 的自動路由。

請求從流量入口進來經由 GNS 和 GSL 尋址，肯定服務所在的 DCN 後，總線會將請求自動路由到對應服務所在的 DCN 區域，交由應用處理。每一個 DCN 內的應用只處理本 DCN 內的請求。應用會訪問同 DCN 內預先分配的主 DB ， DB 層會有一個多副原本提升可靠性。

爲了提高服務的可用性和可靠性， DeFiBus 的開發者針對「 Request-Reply 」的使用作了多個方面的優化和改造。

快速失敗和重試

圖3.2 快速失敗和重試示意圖

從使用方視角來看，業務的超時時間等同於整個完整 RR 調用的超時時間。一次RR調用內部會涉及 2 次消息的發送，當 Broker 有故障時，可能會出現消息發送超時。所以，內部發送消息的超時時間設置會根據業務超時時間自動調整爲較小的值，爲失敗重試留足更多的時間。好比業務超時時間爲 3s ，則設置發送消息的超時爲 1s 。經過調整消息發送超時時間來快速發現 Broker 的故障。當發現 Broker 的故障後， Producer 會當即重試另一個 Broker ，並隔離失敗的 Broker 。在隔離結束前， Producer 不會再將消息發到隔離的 Broker 上。

熔斷機制

圖3.3 熔斷示意圖

熔斷機制是指當某個隊列消息堆積達到指定閾值後，再也不往這個隊列發送消息，使得這個隊列對應的服務實例暫時熔斷。

爲了實現熔斷機制，隊列增長了「隊列深度」屬性。隊列深度指一個隊列中堆積在 Broker 上未被 Consumer 拉取的消息量。當 Consumer 發生故障或者處理異常，首先觸發客戶端的流控機制，隨後拉消息請求會被不斷地延遲，此時消息會堆積在 Broker 上。當 Broker 發現某個隊列堆積的消息量超過閾值，會標記隊列爲熔斷。Producer 發送消息時若是目標隊列已經熔斷，則會收到隊列熔斷的響應碼，並當即重試，發送消息到另外的隊列，同時將熔斷的隊列標記爲隔離。在隔離解除前， Producer 不會再往隔離的隊列發送消息。

隔離機制

隊列級別的隔離機制主要用於 Producer 的重試和服務的熔斷機制。

圖3.4 隔離示意圖

當 Broker 故障時， Consumer 拉消息會觸發隔離機制。原生 RocketMQ 的 Consumer 實現中，由 PullMessageService 單個線程向全部 Broker 發送拉消息請求。當這些 Broker 中有節點故障時， PullMessageService 線程會由於與故障 broker 創建鏈接或者請求響應變慢，致使線程暫時阻塞，這會讓其它正常 Broker 的消息處理耗時變高甚至超時。所以，開發者爲拉消息增長了一個備用線程，一旦發現拉消息的請求執行時間超過閾值，則標記這個 Broker 爲隔離，對應的全部拉消息的請求轉交給備用線程執行，保證 PullMessageService 執行的都是正常的 Broker 的請求。經過線程隔離來保證部分 Broker 的故障不會影響 Consumer 實例拉消息的時效。

隊列動態擴容/縮容

隊列動態擴容和縮容目的是保持隊列數和 Consumer 實例數的一致，使得負載均衡後每一個實例消費的隊列數同樣。在 Producer 均勻發送的狀況下，使得 Consumer 實例不會由於分到的隊列數量不同而出現負載不均衡。

擴容/縮容經過動態調整 Topic 配置的 ReadQueueNum 和 WriteQueueNum 來實現。

在擴容時，首先增長可讀隊列個數，保證 Consumer 先完成監聽，再增長可寫隊列個數，使得 Producer 能夠往新增長的隊列發消息。

圖3.5 隊列擴容示意圖

隊列縮容與擴容的過程相反，先縮小可寫隊列個數，再也不往即將縮掉的隊列發消息，等到 Consumer 將該隊列裏的消息所有消費完成後，再縮小可讀隊列個數，完成縮容過程。

圖3.6 隊列縮容示意圖

負載均衡過渡

RocketMQ Consumer 在負載均衡結果發生變化時，會將老結果直接更新爲新結果，是一個 A 到 B 跳變的過程。當 Consumer 和 Broker 多的時候，不一樣的 Consumer 在負載均衡時獲取到的 Consumer 個數以及隊列個數可能出現不一致，致使負載均衡結果不一致。當結果不一致時就會出現隊列漏聽和重複聽的問題。對於同步調用場景，隊列出現漏聽會致使漏聽隊列上的消息處理耗時變高甚至超時，致使調用失敗。

負載均衡過渡則是將負載均衡結果變化過程增長了一個過渡態，在過渡態的時候， Consumer 會繼續保留上一次負載均衡的結果，直到一個負載均衡週期結束或者感知到新的屬主已經監聽上這個隊列時，才釋放老的結果。

圖3.7 負載均衡過渡示意圖

同城應用多活

爲了達到高可用和容災的一些要求，服務會部署在至少兩個數據中心。當一個數據中心有某個服務所有故障不可用時，其餘數據中心正常的實例能自動接管這部分流量。在部署的時候，請求方和服務方在兩個數據中心都會部署，當兩中心都正常時，請求方會依照服務就近的原則，將請求發到同 IDC 內，跨 IDC 只經過心跳維持鏈接。服務方訂閱時優先監聽同 IDC 內的隊列。

圖3.8 正常狀況示意圖

當且僅當另一個 IDC 中沒有存活的服務實例時，服務方纔會跨 IDC 接管其餘 IDC 的隊列。如圖，當數據中心 2 的應用 B 實例所有掛掉後，部署在數據中心 1 的實例 1 、 2 、3 在負載均衡時首先對同 IDC 內的隊列進行分配，而後檢查發現數據中心 2 有隊列但無存活的應用B實例，此時會將數據中心2的隊列分配給數據中心1的實例，實現跨 IDC 的自動接管。

圖3.9 應用故障狀況示意圖

當某一個數據中心的 Broker 所有掛掉以後，請求方會跨 IDC 進行發送。如圖，在數據中心 2 的 Broker 所有故障後，應用 A 的實例 4~6 會將請求發送到數據中心 1 ，根據服務就近原則，這部分請求會由數據中心 1 的應用 B 實例 1~3 處理，從而保證 Broker 故障後，經由數據中心 2 進來的請求也能被正常處理。

圖3.10 Broker故障狀況示意圖

4、結語

本文主要介紹了 RocketMQ 新特性——「 Request-Reply 」模式。此模式下， Producer 在發出消息後，會等待 Consumer 消費並返回響應消息，達到相似 RPC 調用的效果。「 Request-Reply 」模式讓 RocketMQ 具有了同步調用的能力，在此基礎上，開發者能夠開發更多新的特性。爲了更好的服務於金融場景，微衆銀行又增長了應用多活，服務就近，熔斷等新的特性，構建了安全可靠的金融級消息總線 DeFiBus 。目前微衆銀行已經將大部分紅果經過 DeFiBus 開源出來，後續在分片和尋址方面也會有更通用的實踐總結和成果介紹，歡迎各位瞭解關注！

最後

2019年常見的Java面試題總結了一份將近500頁的pdf文檔，歡迎關注個人公衆號：程序員追風，領取這些整理的資料！

喜歡文章記得關注我點個贊喲，感謝支持！