業界消息總線技術分析-RocketMQ

一、概述

官方簡介：

u? RocketMQ是一款分布式、隊列模型的消息中間件，具有以下特點：

u? 能夠保證嚴格的消息順序

u? 提供豐富的消息拉取模式

u? 高效的訂閱者水平擴展能力

u? 實時的消息訂閱機制

u? 億級消息堆積能力

二、性能吞吐量

單節點，消息大小 10個字節。

Kafka：?? 80萬條/秒

RocketMQ：12萬條/秒

Kafka性能高的原因：producer端多個小消息合并，批量發送到Broker。

可能存在的問題：如果producer宕機，則會導致消息丟失，業務出錯。

三、分布式模型

3.1 網絡部署模型

部署模型：Name server 輕量級的名字服務，存儲cluster、broker、topic、queue之間的關系—即路由信息。

Name Server 是一個幾乎無狀態節點，可集群部署，節點之間無任何信息同步。每個name server都是全量的路由信息。

Broker部署相對復雜，Broker分為Master與Slave，一個Master可以對應多個Slave，但是一個Slave只能對應一個Master，Master與Slave的對應關系通過指定相同的BrokerName，不同的BrokerId來定義，BrokerId=0表示Master，非0表示Slave。Master也可以部署多個。每個Broker與Name Server集群中的所有節點建立長連接，定時注冊Topic信息到所有Name Server。

Producer與Name Server集群中的其中一個節點（隨機選擇）建立長連接，定期從Name Server取Topic路由信息，并向提供Topic 服務的Master建立長連接，且定時向Master發送心跳。Producer 完全無狀態，可集群部署。

Consumer與Name Server集群中的其中一個節點（隨機選擇）建立長連接，定期從Name Server取Topic 路由信息，并向提供Topic服務的Master、Slave建立長連接，且定時向Master、Slave發送心跳。 Consumer既可以從Master訂閱消息，也可以從Slave訂閱消息，訂閱規則由Broker配置決定。

3.2 生產端模型

PUSH（推消息）

單個生產者和該生產者關聯的所有broker保持長連接。

3.3 消費者模型

輪詢消息（PULL模式）

消息拉取線程每隔多久拉取一次？間隔時間由DefaultMQPushConsumer的pullInterval屬性控制，默認為0，可手動設置。

消費完offset存儲到broker上。該時間由DefaultMQPushConsumer的persistConsumerOffsetInterval屬性控制，默認為5秒，可手動設置

四、網絡模型

序號

名稱

描述

優缺點

1

單個 Master

一個集群只有一臺Broker

一旦Broker 重啟或者宕機時，會導致整個服務不可用

2

多 Master 模式

一個集群無 Slave，全是 Master，例如 2 個 Master 或者 3 個 ? Master

優點：配置簡單，單個Master 宕機或重啟維護對應用無影響。

缺點：單臺機器宕機期間，這臺機器上未被消費的消息在機器恢復之前不可訂閱，消息實時性會受到受到影響。

3

多 Master 多 Slave 模式，異步復制

每個 Master 配置一個 Slave，有多對Master-Slave，HA 采用異步復制方式，主備有短暫消息延遲，毫秒級。

優點：即使磁盤損壞，消息丟失的非常少，且消息實時性不會受影響，因為 Master 宕機后，消費者仍然可以從 ? Slave 消費，此過程對應用透明。不需要人工干預。性能同多 Master 模式幾乎一樣。

缺點：Master 宕機，磁盤損壞情況，會丟失少量消息。

4

多 Master 多 Slave 模式，同步雙寫

每個 Master 配置一個 Slave，有多對Master-Slave，HA 采用同步雙寫方式，主備都寫成功，向應用返回成功。

優點：數據與服務都無單點，Master宕機情況下，消息無延遲，服務可用性與數據可用性都非常高

缺點：性能比異步復制模式略低，大約低 10%左右，發送單個消息的 RT 會略高。目前主宕機后，備機不能自動切換為主機，后續會支持自動切換功能。

五、消息模型

隊列模型

Topic 可以有N個隊列組成。隊列可以在一臺Broker上，也可以在不同Broker上。相當于kafka的分區。

Tag：消息的二級分類?？梢酝ㄟ^此進行過濾消費。

消息模型：

序號

名稱

說明

1

順序消息

局部順序，即一類消息為滿足順序性，必須Producer單線程順序發送，且發送到同一個隊列，這樣Consumer就可以按照Producer發送的順序去消費消息。

2

普通順序消息

正常情況下可以保證完全的順序消息，但是一旦發生通信異常，Broker重啟，由于隊列總數發生變化，哈希取模后定位的隊列會變化，產生短暫的消息順序不一致。

如果業務能容忍在集群異常情況（如某個Broker宕機或者重啟）下，消息短暫的亂序，使用普通順序方式比較合適。

3

嚴格順序消息

順序消息的一種，無論正常異常情況都能保證順序，但是犧牲了分布式Failover特性，即Broker集群中只要有一臺機器不可用，則整個集群都不可用，服務可用性大大降低。

目前已知的應用只有數據庫binlog同步強依賴嚴格順序消息，其他應用絕大部分都可以容忍短暫亂序，推薦使用普通的順序消息。

4

事務消息

阿里云MQ支持分布式事務消息，未來開源版本的RocketMQ也有計劃支持分布式事務消息。

消費模型：

序號

名稱

說明

1

廣播消費

一條消息被多個Consumer消費，即使這些Consumer屬于同一個Consumer ? Group，消息也會被Consumer Group中的每個Consumer都消費一次，廣播消費中的Consumer Group概念可以認為在消息劃分方面無意義。

2

集群消費

一個Consumer ? Group中的Consumer實例平均分攤消費消息。例如某個Topic有9條消息，其中一個Consumer Group有3個實例（可能是3個進程，或者3臺機器），那么每個實例只消費其中的3條消息。

3

并發消費

消費者個數比隊列個數多時，并發消費吞吐量大。注意：該情況下，不保序。（這個就相當于我們后面做的并發消費。）

六、消息轉發實時性

(1). Producer 發送消息，消息從socket 進入java 堆。

(2). Producer 發送消息，消息從java 堆轉入PAGACACHE，物理內存。

(3). Producer 發送消息，由異步線程刷盤，消息從PAGECACHE 刷入磁盤。

(4). Consumer 拉消息（正常消費），消息直接從PAGECACHE（數據在物理內存）轉入socket，到達consumer，不經過java 堆。這種消費場景最多，線上96G 物理內存，按照1K 消息算，可以在物理內存緩存1 億條消息。

(5). Consumer 拉消息（異常消費），消息直接從PAGECACHE（數據在虛擬內存）轉入socket。

(6). Consumer 拉消息（異常消費），由于Socket 訪問了虛擬內存，產生缺頁中斷，此時會產生磁盤IO，從磁盤Load 消息到PAGECACHE，然后直接從socket 發出去。

(7). 同5 一致。

(8). 同6 一致。

實時消息消費：大部分都是可以實時消息PageCache的，實時性高。

堆積消息的消費：“讀寫分離設計”，如果Broker有Slave，消費時，當Master發現，消費者消費的是磁盤上的數據，會把該消費者重定向到Slave節點進行讀取。

七、持久化

刷盤：同步/異步刷盤

Replication：同步/異步，參加第四章節。

所有的Topic的隊列寫入同一個CommitLog，每個CommitLog文件默認大小1G，超過1G自動生成新的。

u? 永遠一個文件在寫，其他文件在讀

u? 順序寫，隨機讀

u? 消費時，使用mmap + write方式

消息清理

掃描間隔

默認10秒，由broker配置參數cleanResourceInterval決定

空間閾值

物理文件不能無限制的一直存儲在磁盤，當磁盤空間達到閾值時，不再接受消息，broker打印出日志，消息發送失敗，閾值為固定值85%

清理時機

默認每天凌晨4點，由broker配置參數deleteWhen決定；或者磁盤空間達到閾值

文件保留時長

默認72小時，由broker配置參數fileReservedTime決定

八、消息QoS機制

u? 可靠性

Master slave模式時，數據可靠。可以配置1個master，多slave即多副本。

u? 安全

通信基于Netty的，可以配置SSL通道。

數據存儲安全，開源版本沒有看到。

u? 時間約束

？

u? 消息傳遞的優先級

沒有。優先級會嚴重影響隊列性能。在內存的數據，排序還好點，已經落盤的排序，有點不可能了。

九、擴展性

擴容：

Broker可以自由的擴容。這樣整個集群能力增強，但是對于已經存在的topic的隊列，不能自動rebalance到新增的上去。需要專門工具，開源版本沒有。

Name server 的擴容，Broker怎么動態發現最新的Name server？客戶端（Producer/Comsumer）怎么發現添加的Name server？

?? 環境變量指定NameServer地址·

export NAMESRV_ADDR=192.168.8.106:9876

?? http靜態服務器尋址

客戶端啟動后，會定時訪問一個靜態的HTTP服務器，地址如下：

http://examle.com:8080/rocketmq/msaddr

這個URL的返回內容如下：

192.168.8.106:9876

客戶端默認每隔2分鐘訪問一次這個HTTP服務器，并更新本地的NameServer地址。URL已經在代碼中寫死，可通過修改/etc/hosts文件來改變要訪問的服務器，例如在/etc/hosts增加如下配置：

10.232.22.67?? examle.com

縮容：開源版本沒有更多介紹。

升級：v3.2.6新版本向前向后兼容，客戶端與服務器不同版本可互相兼容。

消息豐富的過濾機制。可擴展。

?? 簡單過濾：

訂閱消息時，可以根據tag過濾。

consumer.subscribe("TopicTest1", "TagA || TagC || TagD");

?? 高級過濾：

可以在Broker上開啟 Filter Server 進程進行過濾。Filter Server上運行的代碼，可以自定義。使用CPU 資源來換取網卡流量資源（因為CPU通常不高，如果客戶端過濾，則浪費帶寬。但是過濾會帶來時延）。

九、總結

Broker部署復雜：Broker本身分為Master/Slave角色

Topic Replica只能按照上面Broker的Master/Slave配置，即如果topic創建在1個master/1個slave上，則具有1個副本。如果創建3副本的topic，則需要專門放在1個master/2 slave的Broker上。不靈活。

但是這種結構比kafka可能更可以支撐大集群。因為

1、 沒有統一的controller節點。

2、 Broker之間的通信也是固定的，master只和自己的slave通信，進行數據的副本復制。而Kafka的Broker需要和不定數目的Broker進行通信。（topic A 副本可能分布在Broker 1，Broker2，Broker 3上，Topic B的副本可能分布在Broker 1 ，Broker 10，Broker20上）

3、? RocketMQ 2.0 是zookeeper，RocketMQ是自研的name server。可以深入思考一下，為什么？name server代碼行數不到1000行，就是個內存K/V數據庫。非常輕量級，沒有watch機制，都是通過心跳檢測。Name server之間不通信，不同步。多臺name server相當于多臺熱備。重新添加一臺name server，每30s Broker更新路由信息，可以全量獲取到所有的信息。

4、RocketMQ支持豐富的消費過濾，定時發送，失敗重傳，事務消息，可能是和電商或者說阿里本身的場景有關，即他們需要這些豐富的功能。我們的核心場景是什么？要有所取舍。

分布式消息服務

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。