大量業務使用消息中間件進行系統間的解耦、異步化、削峰填谷設計實現。公司內部前期基于RabbitMQ實現了一套高可用的消息中間件平臺。隨著業務的持續增長，消息體量隨之增大，對消息中間件平臺提出了更高的要求，此外在運維過程中也遇到了高可用難以保障，功能特性不足等諸多問題。基于遇到的這些問題，決定引入RocketMQ進行替換。本文將介紹基于RocketMQ建設消息中間件平臺并實現在線業務無感知的平滑遷移。
一、背景說明
vivo互聯網中間件團隊于2016年開始基于開源RabbitMQ向業務提供高可用消息中間件平臺服務。
為解決好業務流量快速增長的問題，我們通過合理的業務集群拆分和動態調整， 較好的交付了業務對消息中間件平臺的 平臺能力需求。
但是隨著業務長周期的迅猛發展，消息體量也越來越大，在高并發、大流量場景下RabbitMQ的系統架構設計存在著一定的限制，主要有以下問題：
1.1 高可用能力不足
架構設計存在腦裂風險，并且默認腦裂后無法自動恢復，人工介入恢復存在數據丟失的風險。
為解決腦裂問題，可以選擇將網絡異常后的處理調整為pause_minority模式，但是也帶來了可能微小的網絡抖動也會導致集群故障無法恢復的問題。
1.2. 性能不足
業務消息發送后通過exchange路由到對應的queue中，每一個queue由集群中的某個節點實際承載流量，高流量下集群中的某個節點可能會成為瓶頸。
queue由某個節點承載流量后無法快速遷移，強制遷移到其它低負載節點可能會導致queue不可用，這也導致了向集群中添加節點并無法快速提升集群的流量承載能力。
集群性能較低，經測試使用三臺機器組成集群，可承載大概數萬tps左右，并且由于queue是由集群中某個節點實際承載的，也無法繼續提升某個queue的性能，這樣就無法支撐大流量業務。
消息堆積到千萬或更多后會導致集群性能下降，甚至海量堆積后如果消費請求tps特別高，可能會因為磁盤的性能損耗導致發送性能下降，并且在消息堆積太多時恢復時間長甚至無法恢復。
1.3 功能特性不足
RabbitMQ 默認情況下消費異常會執行立即重新投遞，少量的異常消息也可能導致業務無法消費后續消息。
功能特性上未支持事務消息、順序消息功能。
雖可自行實現消息軌跡邏輯，但是會對集群產生非常大的性能損耗，在正式環境中實際無法基于RabbitMQ原生的能力實現消息軌跡功能。
二、消息中間件平臺的項目目標
基于以上問題，中間件團隊于2020年Q4開始進行了下一代消息中間件平臺方案的調研，為保證下一代消息中間件平臺符合業務新的需求，我們首先明確了消息中間件平臺的建設目標，主要包含兩部分：
- 業務需求
- 平臺需求
2.1 業務需求分析
高性能：可支撐極高的tps，并且支持水平擴展，可快速滿足業務的流量增長需求，消息中間件不應成為業務請求鏈路性能提升的瓶頸點。
豐富的功能特性：支持集群、廣播消費；支持事務消息、順序消息、延時消息、死信消息；支持消息軌跡。
2.2 平臺運維需求分析
-
可運維：業務使用權限校驗；業務生產消費流量限制；業務流量隔離與快速遷移能力。
-
可觀測：豐富的性能指標觀察集群的運行情況。
-
可掌握：可基于開源組件快速進行二次開發，豐富平臺功能特性和進行相關問題修復。
-
云原生：后續可基于容器化提供云原生消息中間件，提供更高的彈性和可伸縮能力。
- 總結： 需要建設高性能、高可靠的下一代消息中間件，具備極高的數據可靠性，豐富的功能特性，并且需要完美兼容當前的RabbitMQ平臺，幫助業務快速遷移到新消息中間件平臺，減少業務遷移成本。
三、開源組件選型調研
基于當前RabbitMQ平臺的問題和對下一代消息中間件平臺的項目需求，我們開展了針對當前較流行的兩款消息中間件：RocketMQ、Pulsar的調研。
調研過程中主要針對以下兩方面進行對比：
3.1 高可用能力分析對比
3.1.1 高可用架構與負載均衡能力對比
Pulsar部署架構（來源：Pulsar社區 ）
RocketMQ部署架構（來源：RocketMQ社區 ）
- Pulsar：
- 采用計算與存儲分離架構設計，可以實現海量數據存儲，并且支持冷熱數據分離存儲。
- 基于ZK和Manager節點控制Broker的故障切換以實現高可用。
- Zookeeper采用分層分片存儲設計，天然支持負載均衡。
RocketMQ：
- 采用存算一體架構設計，主從模式部署，master節點異常不影響消息讀取，Topic采用分片設計。
- 需要二次開發支持主從切換實現高可用。
- 未實現Broker的自動負載均衡，可以將top n流量Topic分布到不同的Broker中實現簡單的負載均衡。
3.1.2 擴縮容與故障恢復對比
- Pulsar
- Broker與BooKeeper獨立擴縮容，并且擴縮容后會完成自動負載均衡。
- Broker節點無狀態，故障后承載Topic會自動轉移到其它Broker節點，完成故障秒級恢復。
- BooKeeper由自動恢復服務進行ledger數據對齊，并恢復到設置的QW份。
- 故障期間已ack消息不會丟失，未ack消息需要客戶端重發。
- RocketMQ
- Broker擴縮容后需要人工介入完成Topic流量均衡，可開發自動負載均衡組件結合Topic的讀寫權限控制自動化完成擴縮容后的負載均衡。
- 基于主從切換實現高可用，由于客戶端定期30秒從NameSrv更新路由，因此故障恢復時間在30~60秒，可以結合客戶端降級策略讓客戶端主動剔除異常Broker節點，實現更快故障恢復。
- 采用同步復制異步刷盤部署架構，在極端情況下會造成少量消息丟失，采用同步復制同步刷盤，已寫入消息不會丟失。
3.1.3 性能對比
Pulsar
- 可支撐百萬Topic數量，實際受到ZK存儲元數據限制。
- 根據內部壓測1KB消息可支撐TPS達數十萬。
RocketMQ
- 邏輯上可支撐百萬Topic，實際在達到數萬時Broker與NameSrv傳輸心跳包可能超時，建議單集群不超過5萬。
- 根據壓測可支撐1KB消息體TPS達10萬+。
3.2 功能特性對比
3.3 總結
從高可用架構分析，Pulsar基于Bookeeper組件實現了架構的計算與存儲分離，可以實現故障的快速恢復；RocketMQ采用了主從復制的架構，故障恢復依賴主從切換。
從功能特性分析，Pulsar支持了豐富的過期策略，支持了消息去重，可以支持實時計算中消息只消費一次的語義；RocketMQ在事務消息、消息軌跡、消費模式等特性對在線業務有更好的支持。
從這兩方面對比，最終選擇了 RocketMQ構建我們下一代的消息中間件平臺。
四、平滑遷移建設
通過技術調研，確定了基于RocketMQ建設下一代消息中間件平臺。
為了實現業務從RabbitMQ平滑遷移到RocketMQ，就 需要建設消息網關實現消息從AMQP協議轉換到RocketMQ；RabbitMQ與RocketMQ的元數據語義與存儲存在差異，需要實現元數據語義的映射與元數據的獨立存儲。
主要有以下四個事項需要完成：
4.1 消息網關獨立部署與嵌入式部署差異對比
4.2 元數據定義映射與維護
4.3 互不干擾的高性能消息推送
RabbitMQ采用推模式進行消息消費，雖然RocketMQ也支持消息推送消費，但是因為AMQP協議中通過prefetch參數限制了客戶端緩存消息數量以保證不會因緩存太多消息導致客戶端內存異常，因此在消息網關實現消息推送時也需要滿足AMQP協議的語義。
同時每個消息網關都需要數千甚至數萬的queue的消息推送，每個queue消息消費速率存在差異，并且每個隊列可能隨時有消息需要推送到客戶端進行消費，要保證不同queue之間的推送互不干擾且及時。
為了實現高效的、互不干擾的消息推送，有以下策略：
- 每個queue采用獨立的線程，保證互不干擾和時效性，缺點是無法支撐海量queue的消息推送。
- 基于信號量、阻塞隊列等，在感知到有可推送消息和可消費服務端時按需進行消息的推送，這樣可使用少量的線程即可完成高效的消息推送。
最終選擇了第2種方案，數據流轉圖如下圖所示：
一個消息消費過程：客戶端在啟動連接到消息網關后，在消息網關中會構建RocketMQ推送消費客戶端實例，并且注入自定義的ConsumeMessageService實例，同時使用一個信號量保存客戶端允許推送的消息數量。
當消息從集群側推送到消息網關時，將消息按照推送的批次封裝為一個任務保存在ConsumeMessageService實例的BlockingQueue中，同時推送線程會輪詢所有的ConsumeMessageService實例，如果發現本地緩存有待消費的消息并且有可消費消息的業務客戶端，將任務提交到線程池中完成消息的推送。
為了保證不會因為少量消費速率特別高的queue導致其它queue的消息推送時效性降低，會限制每一個ConsumeMessageService只允許推送一定數量的消息即轉到推送其它queue的消息，以此即可保證所有queue的消息推送的互不干擾和時效性。
在客戶端消費ack/uack后再次通過信號量通知下一次推送，這樣也保證了使用少量的線程資源即可完成海量消息的推送需求。
4.4 消費啟停與消費限流能力實現
基于消息網關，可以在消息推送邏輯中增加消費啟停和消費限流邏輯。
消費啟停可以幫助業務快速實現消費的暫停或是部分異常節點停止消息消費。
消費限流可以幫助業務控制消息消費速率，避免對底層依賴產生太大壓力。
4.5 平臺架構
- 最終形成了以上的平臺架構。新建設了一個AMQP-proxy消息網關服務實現AMQP消息轉換到RocketMQ，支持業務的消息生產消費。
- 建設了mq-meta服務維護集群的元數據信息。
- 通過mq-controller控制集群的主從切換，實現集群的高可用，同時增加了集群監控，負載均衡模塊保障集群的高可用。
五、平臺建設進展與遷移收益
5.1 業務使用收益
5.1.1 更高、更穩定的消息發送性能
原生RabbitMQ集群業務壓測性能
使用消息網關后業務壓測性能
5.1.2 更豐富的功能特性
- 統一的消息過期時間
- 消費異常消息將按照梯度延時重投遞
- 直接支持廣播消費模式
- 全環境按需提供消息軌跡功能
- 支持消費重置到以前的某個位點
5.1.3 業務使用特性變化
- 消息將 不再無限期保留 ，默認 保留3~7天 （實際保留時間根據集群配置決定）
- 消費異常將不再立即重投遞，將按照一定的 梯度延時重投遞 ，多次異常后將變為死信消息
- 直接支持 廣播消費 ，注意廣播消費模式消費無異常重投遞，每個消息每個節點只消費一次
- 業務生產消費性能可支持水平擴展
- 不支持 消費優先級 功能
- 默認 消費超時時間15分鐘 ，消費超時后消息重新投遞，消費超時時間可按需調整
- 支持 消費啟停 （全局或限制部分節點消費）
- 支持 全局消費限流
- 限制消息體大小 ，當前限制為256KB，超過將直接返回失敗，后續將進行流量治理，限制發送大消息體業務流量
5.2 平臺運維收益
業務從RabbitMQ遷移到RocketMQ后，可支撐業務流量從萬TPS級別提升到十萬TPS級別，可支撐業務容量從數億提升至百億級別。耗用機器資源下降50%以上，運維難度和成本均大大降低，同時可以基于消息網關實現更加豐富的功能特性。
六、未來展望
未來，中間件團隊計劃在三個方面對消息中間件進行迭代演進：
- 基于消息網關能力豐富現有平臺功能特性，進行業務消息治理。
- 過去五年中間件團隊基于開源RabbitMQ進行了RabbitMQ的高可用建設，發現直接讓業務方使用基于開源組件的SDK接入會帶來SDK升級困難，與后端消息中間件類型綁定的問題，未來我們計劃基于GPRC和消息網關，實現消息隊列引擎服務化，業務無需關心底層具體使用的開源消息中間件選型。
- 調研RocketMQ5.0計算與存儲分離構架，進行消息中間件架構的再升級。
END
