我們為什么需要消息總線?

在消息總線上線前，馬蜂窩大部分業務中的異步需求是通過 Redis 隊列來實現。隨著消息量增加，經常會出現消息積壓、不同消息之間互相影響的問題。為解決這些問題，電商研發團隊開始規劃和設計消息總線。

為什么會有消息總線，而不是讓業務系統直接用 PHP 或者其他語言對接 RabbitMQ，Kafka 這樣的消息系統?

「消息總線和直接使用消息系列有什么實際的區別?」，這是很多研發同學一開始不太理解的地方。假如只是為了用一個性能更好的消息系統代替 Redis，確實并不需要消息總線的這個角色。

但當我們從實際業務角度出發，對公司整體技術架構和開發場景的梳理時，發現如果直接讓業務系統對接消息系統，并不是一個很好的方式，并且至少會面臨以下問題：

系統分散。各開發團隊需要維護各自的消息服務，彼此之間相對隔離。

增加開發難度。用戶需要關注具體消息所在消息服務的配置，關注不同業務的消息可能要對接不同種類的消息系統。

維護成本高。用戶需要管理自己消費服務的穩定性，處理各種服務異常，保證消費的可靠性。特別對于 PHP 來說，這個成本還是比較高的。

管理難度大。沒法對業務消息的創建和訂閱關系進行統一管理，也不方便對業務消息中的敏感數據進行權限管理。

不易擴展。無法統一消息系統擴展功能(路由、延時、重試、消費確認等)的使用。

總體來說，直接使用消息系統可以被看成是一個面向技術的接入方式;而消息總線則期望通過隱藏部署、分組和通信等細節，實現一個面向業務的接入方式。

架構設計和技術實現

1. 架構設計

消息總線隱藏了消息發送、路由、分組、存儲、消費負載、通信、高可用等一些列問題。對使用者來說，只需要在發送端調用一個 SDK 消息發送方法，在消費端提供一個 PHP 消費方法即可。

圖1 馬蜂窩消息總線架構設計

馬蜂窩消息總線當前使用 RabbitMQ 作為消息引擎，在發送端提供了 SDK，作為消息總線的 Broker 角色，包含了消息路由分組的功能，負責消息的 Publish。

消息的訂閱關系，目前是持久化在 MySQL 中，在消息發送時會根據訂閱關系把消息投遞到對應的業務消費者。

而在消費端，并沒有直接用 PHP 去接入 RabbitMQ，而是使用 Deliver 服務集群 (Golang 服務) 來負責把 AMQP 協議轉為 HTTP 協議，然后通過 PHPService 進行消費 PHP 代碼的執行。

這個方案在設計時，同時考慮到了未來系統規模擴展后的消息分組，以及關鍵環節的可替代性。

• SDK 充當了消息服務 Broker 的角色，可以控制消息的路由、分組。未來在微服務體系中可以保持整體架構不變，只采用其他方案實現 Broker。
• 可以根據業務場景對接不同的消息引擎，比如對業務一致性要求高的業務使用 RabbitMQ，而對并發要求較高的可以使用 Kafka。對業務來說是無感知的。
• Deliver 和 Application Service 之間可擴展更多的通信協議，支持應用更靈活的消費方式，包括支持未來在微服務中的消費服務。

2. 技術實現

1). 減少流轉復雜度

為了保證消息在消息總線內各環節流轉時減少復雜度，能夠被統一處理，消息體被設計為統一的結構。主要分為以下 3 個部分：

圖2 消息體的定義

• Parameter——參數部分包含消息 ID，來源，時間等參數。
• Conetent——消息內容，在 PHP 中使用者可以把消費方法的輸入參數放入 Content 中。
• Receiver——標注了消息的接收者 (PHP 中為消費者的方法)。

2). 在線服務異步

點對點模式是業務中常用的一種異步模式，

圖3 點對點消息模式

業務應用把不需要在同步請求中執行的邏輯放到異步去執行。發送消息的業務需要明確處理消息的接收者 (消費的 PHP 方法)。消息在發送時需要明確指定唯一的一個 Receiver。

當前通過消息總線 SDK 提供的 invoke 方法可以指定消費的應用方法。

3). 解耦

消息的發布訂閱模式，是一個標準的業務解耦模式。

圖4 發布訂閱(廣播)

App 1 的應用只負責發出消息，至于什么業務需要關注，下游業務應用自己訂閱該消息就可以。很大程度上減少了上游業務和下游業務的耦合程度和開發調試成本。

消息總線使用 DB 來進行消息訂閱關系的存儲，上游業務的消息經過消息總線 Broker 時會根據訂閱關系，裂變為 Receiver 是訂閱應用的多條消息。這樣的消息裂變方式使消息后續在消息總線流轉時目標明確，在進行消費負載，消費確認，失敗重試等場景時可以按照 Receiver 進行隔離。

同樣調用方可以使用 SDK 提供的 pub 方法進行消息的發送，訂閱方通過消息管理系統進行消息訂閱的申請。

4). 防消息干擾

很多使用消息總線的同學比較關心不同消息之間是否會相互干擾，比如由于某個消息短時間內大量涌入是否會造成其他消息被阻塞。

通過前面架構的介紹，可以看到所有的消息經過 Broker 時可以進行路由、分組。消息總線未來會根據業務和消息量來做一些物理隔離，保障業務之間不會相互影響。

而在一個分組內，消息總線也有一些機制保障分組內的不同消息不會相互影響。

圖5 防消息干擾機制

消息經過 Broker 默認會進入一個 Online Queue 的隊列中，Deliver 集群中會有多個 Deliver 監聽 Online Queue。在 Deliver 服務內，通過 Dispatcher 來控制消息總并發消費量，以及同類型消息的并發消費量。當某種類型的并發消息數量超過閾值時，就會被轉發到 Offline Queue，避免消費 Worker 都被同一個類型的消息占用。而 Offline Queue 會被獨立的 Deliver 服務監聽進行消費，不影響 Online Queue 的消費。

5). 消費服務高可用

為了保證消費時的高可用，Deliever 群在負責進行消費協議轉換之外，也做了一些策略來保證消費端的高可用。

• 熔斷

在消息一段時間內失敗數量超過閾值時，停止對隊列的消費，避免由于服務抖動和線上故障引起的大面積消息。

• 消費失敗

熔斷后，Deliver 服務會對后端應用服務健康度進行監控，在服務恢復后可自動恢復消費。

• 系統失敗重試

消息總線服務發生故障時，可對期間的失敗消息采用重試策略進行重試，避免由于基礎服務問題造成的消費失敗。

• 業務失敗重試

在業務應用消費時產生業務異常，可在訂閱消息時指定是否進行重試。消息總線會對需要失敗的消息按照一定的時間周期進行多次重試。

• Graceful 重啟

Deliver 實現了 Graceful 重啟和退出，保障當前正在消費的消息都處理完成后才會進程退出。

未來規劃

圖7 未來演進方向

1. 產品化

當前消息總線在功能上經過近一年的迭代，已經基本穩定。但在消息管理，監控，統計等環節對開發者來說還不夠友好，接下來一段時間會著重優化系統的易用性。

目前已經在規劃以下方向的優化改進：

開發者可以通過消息管理系統進行新增消息，訂閱消息 (加入權限的審核) 等操作，代替當前手工提 issue 的方式。

開發者可以通過系統關注到自己消息的消費情況，并及時接收到消息處理異常的報警。

完善監控體系，提供更精細維度的系統監控數據。

2. 微服務

關于在微服務架構內提供消息總線服務，也已經在計劃當中。包括在微服務內進行消息發送和使用某個微服務進行消息的消費。未來整個消息總線計劃會往下圖的架構進行演進，增加對多語言和不同架構服務的支持。適應更多的業務開發場景，提供更穩定，友好的消息總線服務。

另外對消息引擎的技術選型，未來也會考慮接入 Kafka，RocketMQ 等其他消息隊列服務。根據不同業務場景的消息特性，在發布時選擇進入不同的消息隊列服務。比如對可靠性，數據安全性要求高的消息會進入 RabbitMQ，而對高吞吐量的消息可以進入 Kafka。但對消息的發送方和訂閱方來說都可以不用關心這些細節，仍然按照統一的方式進行接入。

馬蜂窩消息總線服務當前也在不斷迭代中，在很多地方還有不少沒有考慮到的問題。歡迎大家多提寶貴意見，您可以掃描下方二維碼訂閱「馬蜂窩技術」更多內容。

本文作者：梁亮，馬蜂窩電商研發團隊技術專家。2004 年畢業于西安郵電大學，曾在新浪，開心網，阿里巴巴等公司工作。先后從事搜索，社交，視頻，電商等多個方向的研發工作。于 2017 年加入馬蜂窩，現負責馬蜂窩電商平臺服務系統開發。

【本文是51CTO專欄作者馬蜂窩技術的原創文章，作者微信公眾號馬蜂窩技術(ID：mfwtech)】

戳這里，看該作者更多好文