RocketMQ作為阿里系的一款開源的MQ中間件,經歷了雙十一的高并發場景的消息流轉,能夠處理萬億級別的消息。
這篇文章將作為《RocketMQ 進階》專欄的第一篇文章,介紹一下實際生產中如何搭建一個高可用的RocketMQ集群。集群整體架構圖如下:

為什么要用MQ?
消息隊列是一種“先進先出”的數據結構

其應用場景主要包含以下3個方面
1、應用解耦
系統的耦合性越高,容錯性就越低。以電商應用為例,用戶創建訂單后,如果耦合調用庫存系統、物流系統、支付系統,任何一個子系統出了故障或者因為升級等原因暫時不可用,都會造成下單操作異常,影響用戶使用體驗。

使用消息隊列解耦合,系統的耦合性就會提高了。比如物流系統發生故障,需要幾分鐘才能來修復,在這段時間內,物流系統要處理的數據被緩存到消息隊列中,用戶的下單操作正常完成。當物流系統回復后,補充處理存在消息隊列中的訂單消息即可,終端系統感知不到物流系統發生過幾分鐘故障。

2、流量削峰

應用系統如果遇到系統請求流量的瞬間猛增,有可能會將系統壓垮。有了消息隊列可以將大量請求緩存起來,分散到很長一段時間處理,這樣可以大大提到系統的穩定性和用戶體驗。

一般情況,為了保證系統的穩定性,如果系統負載超過閾值,就會阻止用戶請求,這會影響用戶體驗,而如果使用消息隊列將請求緩存起來,等待系統處理完畢后通知用戶下單完畢,這樣總不能下單體驗要好。
處于經濟考量目的:
業務系統正常時段的QPS如果是1000,流量最高峰是10000,為了應對流量高峰配置高性能的服務器顯然不劃算,這時可以使用消息隊列對峰值流量削峰
3、數據分發

通過消息隊列可以讓數據在多個系統更加之間進行流通。數據的產生方不需要關心誰來使用數據,只需要將數據發送到消息隊列,數據使用方直接在消息隊列中直接獲取數據即可

各種MQ產品的比較
常見的MQ產品包括Kafka、ActiveMQ、RabbitMQ、RocketMQ。

關于MQ技術選型詳細可以看筆者之前的文章:聊聊 MQ 技術選型
RocketMQ中的幾個重要角色
從上述的集群架構圖中可以知道RocketMQ中涉及到的幾個重要的角色:
- NameServer:相當于微服務中的注冊中心,提供broker的服務發現和注冊功能,各個節點之間無相互通信,一旦broker節點啟動,將會主動上報信息給NameServer。
- producer:消息生產者,發送消息給broker;擁有同一個的groupId的producer為一個集群
- broker:消息暫存和傳輸,接收producer發送的消息,采用push/pull模式傳遞給consumer
- consumer:消息消費者,消費broker傳遞的消息,擁有同一個groupId的consumer為一個集群
以上四個是RocketMQ對外四種角色,另外內部還有一些重要角色,如下:
- Topic:消息主題,通過Topic對不同的業務消息進行分類。
- Tag:消息標簽,用來進一步區分某個Topic下的消息分類,消息從生產者發出即帶上的屬性。
- Message Queue:隊列,相當于Topic的分區,用于并行發送和消費消息,一個Topic中對應多個Queue
關于Topic和Tag的區別:比如電商中的下單、支付流程,為了提高并發量通常都會使用消息隊列進行異步處理,那么可以定義消息的Topic為Topic_order,但是其中還涉及了創建訂單、付款、完成訂單這三類消息,如何去區分?
此時就該用到Tag去細分了,此時的對應關系如下圖:

Topic和Message Queue的關系如下圖:

一個Topic中包含多個Message Queue(隊列)
RocketMQ下載
阿里將RocketMQ貢獻給了Apache,所以要去Apache的官網去下載對應的版本;
地址:https://rocketmq.apache.org/dowloading/releases/
我的《RocketMQ 進階》這個專欄選用的版本是4.9.4
下載地址:https://rocketmq.apache.org/download
集群特點
針對RocketMQ對外的四種角色,集群部署有以下幾點需要注意的地方:
- NameServer之間是不相互通信的,因此NameServer集群部署非常簡單,直接啟動多個服務
- broker部署分為master和slave節點,一個master對應多個slave節點,主從節點通過brokerId區分,主節點為0,從節點為1;每個broker與NameServer建立長連接,定時注冊Topic信息到NameServer中。
- producer與NameServer中的任意一個節點建立長連接,定期獲取Topic路由信息,并向提供Topic服務的Master建立長連接,且定時向Master發送心跳。Producer完全無狀態,可集群部署。
- consumer與NameServer集群中的其中一個節點(隨機選擇)建立長連接,定期從NameServer取Topic路由信息,并向提供Topic服務的Master、Slave建立長連接,且定時向Master、Slave發送心跳。Consumer既可以從Master訂閱消息,也可以從Slave訂閱消息,訂閱規則由Broker配置決定。
集群模式
注意這里說的集群模式是針對broker,因為涉及到broker的節點之間的數據同步問題。
NameServer各個節點間不互相通信,只需要啟動多個服務便可實現一個集群
RocketMQ支持四種集群模式,如下:
1. 單Master模式
不建議使用,一旦服務重啟或者宕機將導致整個服務不可用
2. 多Master模式
這個集群模式無slave節點,全部都是master節點,該模式如下圖:

該模式的優缺點如下:
- 優點:該模式性能最高
- 缺點:一旦一臺服務宕機了,那么在這臺服務上的消息不能被訂閱消費,消息實時性會受到影響
3. 多Master多Slave(同步)
每個master對應一個slave節點,有多對master-slave,主從之間的數據復制采用同步雙寫的形式,如下圖:

主從同步雙寫是什么意思?
producer發送一條消息給broker的主節點,只有主節點將數據同步到從節點才會返回結果
此時的發送消息流程如下:

需要經過以上4步才能實現消息發送成功,此時如果主從數據復制阻塞,那么producer必須等待直到成功。
這種模式的優缺點如下:
- 優點:無單點故障,數據不會丟失,即使master宕機了,salve節點依然能夠對外提供服務
- 缺點:由于是同步復制,性能比異步復制的模式低
4. 多Master多Slave(異步)
每個Master配置一個Slave,有多對master-slave,采用異步復制的方式,如下:

消息發送到的master后直接返回,不必等待主從復制,而是內部通過異步的方式進行復制。
該種模式的優缺點如下:
- 優點:無單點故障,消息無延遲,即使master宕機了,salve節點依然能夠對外提供服務
- 性能比異步復制模式略低(大約低10%左右),發送單個消息的RT會略高。
主從同步集群搭建
根據上面的介紹,主從同步集群模式使用4個節點,分別是兩個主節點、兩個從節點。
筆者這里是使用兩臺機器將節點均攤,如下圖:

在安裝之前需要做些準備工作,如下:
- 準備兩臺服務器/虛擬機
- 安裝好JDK1.8的環境
- 下載好rocketmq-all-4.9.4-bin-release
筆者使用的是Centos7的虛擬機進行演示,如下:
序號 | IP | 角色 | 架構模式 |
1 | 192.168.47.146 | nameserver、brokerserver | Master1、Slave2 |
2 | 192.168.47.145 | nameserver、brokerserver | Master2、Slave1 |
1. 添加環境變量
RocketMQ的啟動需要依賴的一個環境變量:ROCKETMQ_HOME(RocketMQ的根目錄)
export PATH=$JAVA_HOME/bin:$PATH
export ROCKETMQ_HOME=/usr/local/rocketmq-all-4.9.4-bin-release
export PATH=$PATH:$ROCKETMQ_HOME/bin
除了以上RocketMQ的環境變量配置,還需添加JDK的配置,省略...
配置保存之后,執行下述命令:
2. 創建消息存儲路徑
RocketMQ是將消息存儲在磁盤,因此需要創建存儲路徑,如下:
mkdir -p /usr/local/rocketmq/store/master
mkdir -p /usr/local/rocketmq/store/master/commitlog
mkdir -p /usr/local/rocketmq/store/master/consumequeue
mkdir -p /usr/local/rocketmq/store/master/index
mkdir -p /usr/local/rocketmq/store/slave
mkdir -p /usr/local/rocketmq/store/slave/commitlog
mkdir -p /usr/local/rocketmq/store/slave/consumequeue
mkdir -p /usr/local/rocketmq/store/slave/index
3. broker配置文件
總共四個節點,分別配置如下:
(1)master1
這個配置文件是broker-a.properties,如下:
#所屬集群名字
brokerClusterName=rocketmq-cluster
#broker名字,注意此處不同的配置文件填寫的不一樣
brokerName=broker-a
#0 表示 Master,>0 表示 Slave
brokerId=0
#nameServer地址,分號分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在發送消息時,自動創建服務器不存在的topic,默認創建的隊列數
defaultTopicQueueNums=4
#是否允許 Broker 自動創建Topic,建議線下開啟,線上關閉
autoCreateTopicEnable=true
#是否允許 Broker 自動創建訂閱組,建議線下開啟,線上關閉
autoCreateSubscriptinotallow=true
#Broker 對外服務的監聽端口
listenPort=10911
#刪除文件時間點,默認凌晨 4點
deleteWhen=04
#文件保留時間,默認 48 小時
fileReservedTime=120
#commitLog每個文件的大小默認1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每個文件默認存30W條,根據業務情況調整
mapedFileSizeCnotallow=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#檢測物理文件磁盤空間
diskMaxUsedSpaceRatio=88
#存儲路徑
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存儲路徑
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消費隊列存儲路徑存儲路徑
storePathCnotallow=/usr/local/rocketmq/store/consumequeue
#消息索引存儲路徑
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存儲路徑
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存儲路徑
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushCnotallow=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 異步復制Master
#- SYNC_MASTER 同步雙寫Master
#- SLAVE
brokerRole=SYNC_MASTER
#刷盤方式
#- ASYNC_FLUSH 異步刷盤
#- SYNC_FLUSH 同步刷盤
flushDiskType=SYNC_FLUSH
#checkTransactinotallow=false
#發消息線程池數量
#sendMessageThreadPoolNums=128
#拉消息線程池數量
#pullMessageThreadPoolNums=128
(2)slave2
修改配置文件broker-b-s.properties,如下:
#所屬集群名字
brokerClusterName=rocketmq-cluster
#broker名字,注意此處不同的配置文件填寫的不一樣
brokerName=broker-b
#0 表示 Master,>0 表示 Slave
brokerId=1
#nameServer地址,分號分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在發送消息時,自動創建服務器不存在的topic,默認創建的隊列數
defaultTopicQueueNums=4
#是否允許 Broker 自動創建Topic,建議線下開啟,線上關閉
autoCreateTopicEnable=true
#是否允許 Broker 自動創建訂閱組,建議線下開啟,線上關閉
autoCreateSubscriptinotallow=true
#Broker 對外服務的監聽端口
listenPort=11011
#刪除文件時間點,默認凌晨 4點
deleteWhen=04
#文件保留時間,默認 48 小時
fileReservedTime=120
#commitLog每個文件的大小默認1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每個文件默認存30W條,根據業務情況調整
mapedFileSizeCnotallow=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#檢測物理文件磁盤空間
diskMaxUsedSpaceRatio=88
#存儲路徑
storePathRootDir=/usr/local/rocketmq/store/slave
#commitLog 存儲路徑
storePathCommitLog=/usr/local/rocketmq/store/slave/commitlog
#消費隊列存儲路徑存儲路徑
storePathCnotallow=/usr/local/rocketmq/store/slave/consumequeue
#消息索引存儲路徑
storePathIndex=/usr/local/rocketmq/store/slave/index
#checkpoint 文件存儲路徑
storeCheckpoint=/usr/local/rocketmq/store/slave/checkpoint
#abort 文件存儲路徑
abortFile=/usr/local/rocketmq/store/slave/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushCnotallow=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 異步復制Master
#- SYNC_MASTER 同步雙寫Master
#- SLAVE
brokerRole=SLAVE
#刷盤方式
#- ASYNC_FLUSH 異步刷盤
#- SYNC_FLUSH 同步刷盤
flushDiskType=ASYNC_FLUSH
#checkTransactinotallow=false
#發消息線程池數量
#sendMessageThreadPoolNums=128
#拉消息線程池數量
#pullMessageThreadPoolNums=128
(3)master2
修改broker-b.properties,如下:
#所屬集群名字
brokerClusterName=rocketmq-cluster
#broker名字,注意此處不同的配置文件填寫的不一樣
brokerName=broker-b
#0 表示 Master,>0 表示 Slave
brokerId=0
#nameServer地址,分號分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在發送消息時,自動創建服務器不存在的topic,默認創建的隊列數
defaultTopicQueueNums=4
#是否允許 Broker 自動創建Topic,建議線下開啟,線上關閉
brokerClusterName=rocketmq-cluster
#broker名字,注意此處不同的配置文件填寫的不一樣
brokerName=broker-b
#0 表示 Master,>0 表示 Slave
brokerId=0
#nameServer地址,分號分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在發送消息時,自動創建服務器不存在的topic,默認創建的隊列數
defaultTopicQueueNums=4
#是否允許 Broker 自動創建Topic,建議線下開啟,線上關閉
autoCreateTopicEnable=true
#是否允許 Broker 自動創建訂閱組,建議線下開啟,線上關閉
autoCreateSubscriptinotallow=true
#Broker 對外服務的監聽端口
listenPort=10911
#刪除文件時間點,默認凌晨 4點
deleteWhen=04
#文件保留時間,默認 48 小時
fileReservedTime=120
#commitLog每個文件的大小默認1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每個文件默認存30W條,根據業務情況調整
mapedFileSizeCnotallow=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#檢測物理文件磁盤空間
diskMaxUsedSpaceRatio=88
#存儲路徑
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存儲路徑
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消費隊列存儲路徑存儲路徑
storePathCnotallow=/usr/local/rocketmq/store/consumequeue
#消息索引存儲路徑
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存儲路徑
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存儲路徑
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushCnotallow=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 異步復制Master
#- SYNC_MASTER 同步雙寫Master
#- SLAVE
brokerRole=SYNC_MASTER
#刷盤方式
#- ASYNC_FLUSH 異步刷盤
#- SYNC_FLUSH 同步刷盤
flushDiskType=SYNC_FLUSH
#checkTransactinotallow=false
#發消息線程池數量
#sendMessageThreadPoolNums=128
#拉消息線程池數量
#pullMessageThreadPoolNums=128
?(4)slave1
修改broker-a-s.properties,如下:
#所屬集群名字
brokerClusterName=rocketmq-cluster
#broker名字,注意此處不同的配置文件填寫的不一樣
brokerName=broker-a
#0 表示 Master,>0 表示 Slave
brokerId=1
#nameServer地址,分號分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在發送消息時,自動創建服務器不存在的topic,默認創建的隊列數
defaultTopicQueueNums=4
#是否允許 Broker 自動創建Topic,建議線下開啟,線上關閉
autoCreateTopicEnable=true
#是否允許 Broker 自動創建訂閱組,建議線下開啟,線上關閉
autoCreateSubscriptinotallow=true
#Broker 對外服務的監聽端口
listenPort=11011
#刪除文件時間點,默認凌晨 4點
deleteWhen=04
#文件保留時間,默認 48 小時
fileReservedTime=120
#commitLog每個文件的大小默認1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每個文件默認存30W條,根據業務情況調整
mapedFileSizeCnotallow=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#檢測物理文件磁盤空間
diskMaxUsedSpaceRatio=88
#存儲路徑
storePathRootDir=/usr/local/rocketmq/store/slave
#commitLog 存儲路徑
storePathCommitLog=/usr/local/rocketmq/store/slave/commitlog
#消費隊列存儲路徑存儲路徑
storePathCnotallow=/usr/local/rocketmq/store/slave/consumequeue
#消息索引存儲路徑
storePathIndex=/usr/local/rocketmq/store/slave/index
#checkpoint 文件存儲路徑
storeCheckpoint=/usr/local/rocketmq/store/slave/checkpoint
#abort 文件存儲路徑
abortFile=/usr/local/rocketmq/store/slave/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushCnotallow=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 異步復制Master
#- SYNC_MASTER 同步雙寫Master
#- SLAVE
brokerRole=SLAVE
#刷盤方式
#- ASYNC_FLUSH 異步刷盤
#- SYNC_FLUSH 同步刷盤
flushDiskType=ASYNC_FLUSH
#checkTransactinotallow=false
#發消息線程池數量
#sendMessageThreadPoolNums=128
#拉消息線程池數量
#pullMessageThreadPoolNums=128
關于上面的各個配置有什么用后面章節會詳細介紹
4、開放端口
宿主機需要遠程訪問虛擬機的rocketmq服務和web服務,需要開放相關的端口號,簡單粗暴的方式是直接關閉防火墻
# 關閉防火墻
systemctl stop firewalld.service
# 查看防火墻的狀態
firewall-cmd --state
# 禁止firewall開機啟動
systemctl disable firewalld.service
或者為了安全,只開放特定的端口號,RocketMQ默認使用3個端口:9876 、10911 、11011 。如果防火墻沒有關閉的話,那么防火墻就必須開放這些端口:
- nameserver 默認使用 9876 端口
- master 默認使用 10911 端口
- slave 默認使用11011 端口
執行以下命令:
# 開放name server默認端口
firewall-cmd --remove-port=9876/tcp --permanent
# 開放master默認端口
firewall-cmd --remove-port=10911/tcp --permanent
# 開放slave默認端口 (當前集群模式可不開啟)
firewall-cmd --remove-port=11011/tcp --permanent
# 重啟防火墻
firewall-cmd --reload
5. Host添加信息
需要在hosts中添加信息,這樣后面的配置就不用通過ip指定了。
執行如下命令進入hosts文件:
配置信息如下:
# nameserver
192.168.47.146 rocketmq-nameserver1
192.168.47.145 rocketmq-nameserver2
# broker
192.168.47.146 rocketmq-master1
192.168.47.146 rocketmq-slave2
192.168.47.145 rocketmq-master2
192.168.47.145 rocketmq-slave1
配置完成后, 重啟網卡:
systemctl restart network
6. 修改啟動腳本
內置RocketMQ啟動對服務器內存要求較高,由于筆者本地測試的配置較低,因此需要修改JVM啟動參數,以下兩個腳本都在bin目錄下。
(1)runbroker.sh腳本修改:

根據自己服務器的配置進行修改
(2) runserver.sh 腳本修改:

7. 服務啟動
RocketMQ啟動分為兩步:
(1)啟動NameServer
分別在兩臺服務器上啟動,命令如下:
cd /usr/local/rocketmq-all-4.9.4-bin-release/bin
nohup sh mqnamesrv &
(2)啟動broker集群
這里master和slave總計四個,均攤在兩個服務器上,下面分別啟動
master1啟動,命令如下:
cd /usr/local/rocketmq-all-4.9.4-bin-release/bin
nohup sh mqbroker -c ../conf/2m-2s-sync/broker-a.properties &
slave2啟動,命令如下:
cd /usr/local/rocketmq-all-4.9.4-bin-release/bin
nohup sh mqbroker -c ../conf/2m-2s-sync/broker-b-s.properties &
master1和slave2在同一臺服務器上(192.168.47.146)
master2啟動,命令如下:
cd /usr/local/rocketmq-all-4.9.4-bin-release/bin
nohup sh mqbroker -c ../conf/2m-2s-sync/broker-b.properties &
slave1啟動,命令如下:
cd /usr/local/rocketmq-all-4.9.4-bin-release/bin
nohup sh mqbroker -c ../conf/2m-2s-sync/broker-a-s.properties &
master2和slave1在同一臺服務器上(192.168.47.145)
8. 查看進程狀態
第7步啟動成功后,查詢進程狀態觀察RocketMQ是否啟動成功,命令如下:

9. 查看日志
同時也可以觀察RocketMQ的日志看下是否異常,命令如下:
# 查看nameServer日志
tail -500f ~/logs/rocketmqlogs/namesrv.log
# 查看broker日志
tail -500f ~/logs/rocketmqlogs/broker.log
集群監控平臺
RocketMQ有一個對其擴展的開源項目rocketmq-dashboard,直接將該項目拉到本地,修改其中的幾個參數編譯打包即可
修改application.yml中的NameServer的配置,改成自己搭建的地址,如下:

然后打包運行,命令如下:
//打包
mvn clean package -Dmaven.test.skip=true
//運行
java -jar target/rocketmq-dashboard-1.0.1-SNAPSHOT.jar
運行成功之后,瀏覽器訪問:http://ip:8080

進入集群這一欄,看下自己搭建的集群信息,如下圖:

總結
本節內容主要介紹了MQ的基本知識以及RocketMQ集群搭建過程,有興趣的可以按照筆者的整個搭建過程嘗試一遍,至于其中一些配置屬性以及生產、消費消息將會在后文介紹。