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示意圖

1、對請求消息增加對應的標識
Producer 發送請求消息時，需要在消息的 Properties 里增加RR調用的標識，其中關鍵的字段有 Correlation_Id、REPLY_TO_CLIENT。Correlation_Id 用來唯一標識一次RR請求，通過這個屬性來匹配同一個RR調用的請求消息和響應消息。REPLY_TO_CLIENT 用來標識請求消息的發出方，其值為 Producer 的 ClientId 。

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

2、發完請求消息后等待響應消息。
請求方每次執行 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故障情況示意圖

四、結語

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

作者信息：陳廣勝，Apache RocketMQ Committer，DeFiBus 創始人，微眾銀行技術專家，中間件平臺相關產品負責人，曾就職于 IBM 和華為，負責運營商云和大數據平臺建設。

原文鏈接
本文為云棲社區原創內容，未經允許不得轉載。

總結

以上是生活随笔為你收集整理的揭秘 RocketMQ 新特性以及在金融场景下的实践的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。