教你一手如何基于RocketMQ搭建生產(chǎn)級(jí)消息集群

前言

目前很多互聯(lián)網(wǎng)公司的系統(tǒng)都在朝著微服務(wù)化、分布式化系統(tǒng)的方向在演進(jìn)，這帶來(lái)了很多好處，也帶來(lái)了一些棘手的問(wèn)題，其中最棘手的莫過(guò)于數(shù)據(jù)一致性問(wèn)題了。早期我們的軟件功能都在一個(gè)進(jìn)程中，數(shù)據(jù)的一致性可以通過(guò)數(shù)據(jù)庫(kù)本地事務(wù)來(lái)加以控制。而在分布式架構(gòu)下，原本比較完整的本地功能可能被拆分成了多個(gè)獨(dú)立的服務(wù)進(jìn)程。與之前相比，同樣一筆業(yè)務(wù)訂單此時(shí)可能會(huì)經(jīng)歷很多服務(wù)模塊的處理，調(diào)用鏈路會(huì)變得很長(zhǎng)，例如某電商平臺(tái)，一筆購(gòu)物訂單可能會(huì)經(jīng)過(guò)：商品中心、訂單、支付、物流等多個(gè)服務(wù)的調(diào)用，而這可能還只是比較粗粒度的劃分，某些比較大型的服務(wù)，如支付系統(tǒng)，可能本身又會(huì)按照分布式的架構(gòu)拆分成多個(gè)微服務(wù)，所以整個(gè)業(yè)務(wù)的調(diào)用鏈路會(huì)變得更加冗長(zhǎng)。

而這不可避免的就會(huì)產(chǎn)生數(shù)據(jù)不一致的問(wèn)題，為了實(shí)現(xiàn)業(yè)務(wù)上的最終一致性，然后功能比較獨(dú)立的系統(tǒng)，如訂單系統(tǒng)與支付系統(tǒng)就會(huì)通過(guò)額外的業(yè)務(wù)邏輯設(shè)計(jì)來(lái)確保彼此之間的最終一致性，如訂單系統(tǒng)會(huì)通過(guò)訂單的支付狀態(tài)來(lái)保持與支付系統(tǒng)的數(shù)據(jù)一致，而支付系統(tǒng)則會(huì)提供支付狀態(tài)查詢接口，或者實(shí)現(xiàn)最大可能的主動(dòng)回調(diào)功能，來(lái)確保二者數(shù)據(jù)狀態(tài)的最終一致。此外可能還會(huì)通過(guò)日終的訂單對(duì)賬來(lái)發(fā)現(xiàn)不一致的數(shù)據(jù)，并進(jìn)行數(shù)據(jù)校正。

但是這些都只是業(yè)務(wù)邏輯上的手段，對(duì)于某些內(nèi)部服務(wù)之間的調(diào)用，如果可以通過(guò)分布式事務(wù)解決方案來(lái)加以保證的話，其實(shí)是可以大大減少一些不必要的復(fù)雜業(yè)務(wù)邏輯的。實(shí)際上，目前市面上能夠提供分布式事務(wù)解決方案、又比較成熟的開(kāi)源技術(shù)框架比較少，而RocketMQ在4.3.0之后的版本提供了事務(wù)消息的功能，因?yàn)镽ocketMQ本身?yè)碛斜容^多的生產(chǎn)實(shí)踐的關(guān)系，所以這一功能備受關(guān)注，作者所在的公司也有一些實(shí)踐。

以此為契機(jī)，為了給大家關(guān)于分布式事務(wù)一個(gè)比較清晰的認(rèn)識(shí)，這里我打算以RocketMQ的事務(wù)消息功能為示例，來(lái)相對(duì)全面的總結(jié)下分布式事務(wù)的內(nèi)容。本篇文章的主要內(nèi)容，是先介紹如何搭建一套生產(chǎn)級(jí)的RocketMQ消息集群，以此準(zhǔn)備下試驗(yàn)環(huán)境。

什么是RocketMQ

RocketMQ是阿里開(kāi)源的并貢獻(xiàn)給Apache基金會(huì)的一款分布式消息平臺(tái)，它具有低延遲、高性能和可靠性、萬(wàn)億級(jí)容量和靈活的可伸縮性的特點(diǎn)，單機(jī)也可以支持億級(jí)的消息堆積能力、單機(jī)寫(xiě)入TPS單實(shí)例約7萬(wàn)條/秒，單機(jī)部署3個(gè)Broker，可以跑到最高12萬(wàn)條/秒。

基于以上強(qiáng)大的性能，以及阿里的技術(shù)影響力，RocketMQ目前在國(guó)內(nèi)互聯(lián)網(wǎng)公司中被使用得比較廣泛。那么，我們先大概來(lái)了解一下RocketMQ的整體結(jié)構(gòu)吧！

整個(gè)RocketMQ消息系統(tǒng)主要由如下4個(gè)部分組成：

從中間件服務(wù)角度來(lái)看整個(gè)RocketMQ消息系統(tǒng)（服務(wù)端）主要分為：NameSrv和Broker兩個(gè)部分。

NameSrv：在RocketMQ分布式消息系統(tǒng)中，NameSrv主要提供兩個(gè)功能：

提供服務(wù)發(fā)現(xiàn)和注冊(cè)，這里主要是管理Broker，NameSrv接受來(lái)自Broker的注冊(cè)，并通過(guò)心跳機(jī)制來(lái)檢測(cè)Broker服務(wù)的健康性；

提供路由功能，集群(這里是指以集群方式部署的NameSrv)中的每個(gè)NameSrv都保存了Broker集群(這里是指以集群方式部署的Broker)中整個(gè)的路由信息和隊(duì)列信息。這里需要注意，在NameSrv集群中，每個(gè)NameSrv都是相互獨(dú)立的，所以每個(gè)Broker需要連接所有的NameSrv，每創(chuàng)建一個(gè)新的topic都要同步到所有的NameSrv上。

Broker：主要是負(fù)責(zé)消息的存儲(chǔ)、傳遞、查詢以及高可用(HA)保證等。其由如下幾個(gè)子模塊（源碼總體上也是這么拆分的）構(gòu)成：

• remoting，是Broker的服務(wù)入口，負(fù)責(zé)客戶端的接入（Producer和Consumer）和請(qǐng)求處理。
• client，管理客戶端和維護(hù)消費(fèi)者對(duì)于Topic的訂閱。
• store，提供針對(duì)存儲(chǔ)和消息查詢的簡(jiǎn)單的API（數(shù)據(jù)存儲(chǔ)在物理磁盤(pán)）。
• HA， 提供數(shù)據(jù)在主從節(jié)點(diǎn)間同步的功能特性。
• Index，通過(guò)特定的key構(gòu)建消息索引，并提供快速的索引查詢服務(wù)。

而從客戶端的角度看主要有：Producer、Consumer兩個(gè)部分。

*Producer：消息的生產(chǎn)者，由用戶進(jìn)行分布式部署，消息由Producer通過(guò)多種負(fù)載均衡模式發(fā)送到Broker集群，發(fā)送低延時(shí)，支持快速失敗。

Consumer：消息的消費(fèi)者，也由用戶部署，支持PUSH和PULL兩種消費(fèi)模式，支持集群消費(fèi)和廣播消息，提供實(shí)時(shí)的消息訂閱機(jī)制，滿足大多數(shù)消費(fèi)場(chǎng)景。

來(lái)總結(jié)下，整個(gè)RocketMQ消息集群就是由NameSrv/Broker、Producer/Consumer組成的。為了讓大家更清晰的理解它們之間的關(guān)系，我們以一條完整的信息流轉(zhuǎn)為例，來(lái)看看RocketMQ消息系統(tǒng)是如何運(yùn)轉(zhuǎn)的，如下圖所示：

看到這里相信大家應(yīng)該對(duì)RocketMQ有一個(gè)大致的了解了，那么下面我們就具體看看，如何搭建一套生產(chǎn)級(jí)的RocketMQ消息集群系統(tǒng)吧！

RocketMQ集群模式

RocketMQ集群部署有多種模式，對(duì)于NameSrv來(lái)說(shuō)可以同時(shí)部署多個(gè)節(jié)點(diǎn)，并且這些節(jié)點(diǎn)間也不需要有任何的信息同步，這里因?yàn)槊總€(gè)NameSrv節(jié)點(diǎn)都會(huì)存儲(chǔ)全量路由信息，在NameSrv集群模式下，每個(gè)Broker都需要同時(shí)向集群中的每個(gè)NameSrv節(jié)點(diǎn)發(fā)送注冊(cè)信息，所以這里對(duì)于NameSrv的集群部署來(lái)說(shuō)并不需要做什么額外的設(shè)置。

而對(duì)于Broker集群來(lái)說(shuō)就有多種模式了，這里我先給大家介紹下這幾種模式，然后我們?cè)賮?lái)看看生產(chǎn)級(jí)的部署方式是什么：

1）、單個(gè)Master模式

一個(gè)Broker作為主服務(wù)，不設(shè)置任何Slave，很顯然這種方式風(fēng)險(xiǎn)比較大，存在單節(jié)點(diǎn)故障會(huì)導(dǎo)致整個(gè)基于消息的服務(wù)掛掉，所以生產(chǎn)環(huán)境不可能采用這種模式。

2）、多Master模式

這種模式的Broker集群，全是Master，沒(méi)有Slave節(jié)點(diǎn)。這種方式的優(yōu)缺點(diǎn)如下：

優(yōu)點(diǎn)：配置會(huì)比較簡(jiǎn)單一些，如果單個(gè)Master掛掉或重啟維護(hù)的話對(duì)應(yīng)用是沒(méi)有什么影響的。如果磁盤(pán)配置為RAID10（服務(wù)器的磁盤(pán)陣列模式，遺忘的同學(xué)可以自己查下資料）的話，即使在機(jī)器宕機(jī)不可恢復(fù)的情況下，由于RAID10磁盤(pán)本身的可靠性，消息也不會(huì)丟失（異步刷盤(pán)丟失少量消息，同步刷盤(pán)一條不丟），這種Broker的集群模式性能相對(duì)來(lái)說(shuō)是最高的。

缺點(diǎn)：就是在單臺(tái)機(jī)器宕機(jī)期間，這臺(tái)機(jī)器上未被消費(fèi)的消息在機(jī)器恢復(fù)之前是不可以進(jìn)行消息訂閱的，這對(duì)消息的實(shí)時(shí)性會(huì)有一些影響。

3）、多Master多Slave模式（異步復(fù)制）

在這種模式下Broker集群存在多個(gè)Master節(jié)點(diǎn)，并且每個(gè)Master節(jié)點(diǎn)都會(huì)對(duì)應(yīng)一個(gè)Slave節(jié)點(diǎn)，有多對(duì)Master-Slave，HA(高可用)之間采用異步復(fù)制的方式進(jìn)行信息同步，在這種方式下主從之間會(huì)有短暫的毫秒級(jí)的消息延遲。

優(yōu)點(diǎn)：在這種模式下即便磁盤(pán)損壞了，消息丟失的情況也非常少，因?yàn)橹鲝闹g有信息備份；并且，在這種模式下消息的實(shí)時(shí)性也不會(huì)受影響，因?yàn)镸aster宕機(jī)后Slave可以自動(dòng)切換為Master模式，這樣Consumer仍然可以通過(guò)Slave進(jìn)行消息消費(fèi)，而這個(gè)過(guò)程對(duì)應(yīng)用來(lái)說(shuō)則是完全透明的，并不需要人工干預(yù)；另外，這種模式的性能與多Master模式幾乎差不多。

缺點(diǎn)：如果Master宕機(jī)，并且在磁盤(pán)損壞的情況下，會(huì)丟失少量的消息。

4）、多Master多Slave模式（同步復(fù)制）

這種模式與3）差不多，只是HA采用的是同步雙寫(xiě)的方式，即主備都寫(xiě)成功后，才會(huì)向應(yīng)用返回成功。

優(yōu)點(diǎn)：在這種模式下數(shù)據(jù)與服務(wù)都不存在單點(diǎn)的情況，在Master宕機(jī)的情況下，消息也沒(méi)有延遲，服務(wù)的可用性以及數(shù)據(jù)的可用性都非常高。

缺點(diǎn)：性能相比于異步復(fù)制來(lái)說(shuō)略低一些（大約10%）；另外一個(gè)缺點(diǎn)就是相比于異步復(fù)制，目前Slave備機(jī)還暫時(shí)不能實(shí)現(xiàn)自動(dòng)切換為Master，可能后續(xù)的版本會(huì)支持Master-Slave的自動(dòng)切換功能。

生產(chǎn)級(jí)RocketMQ集群

綜合考慮以上集群模式的優(yōu)缺點(diǎn)，在實(shí)際生產(chǎn)環(huán)境中目前基于RocketMQ消息集群的部署方式基本都是采用多Master多Slave（異步復(fù)制）這種模式，作者目前所在公司的生產(chǎn)環(huán)境的Rocket消息系統(tǒng)也是采用這種模式進(jìn)行部署的。

以下為目前作者所在公司的實(shí)際部署結(jié)構(gòu)：

在以上實(shí)踐中，部署了3個(gè)NameSrv節(jié)點(diǎn)，Broker采用2主2從的異步復(fù)制模式進(jìn)行集群部署。

為了更好地理解RocketMQ的集群運(yùn)行原理，接下來(lái)我們以4臺(tái)虛擬機(jī)來(lái)模擬上述集群的搭建過(guò)程，假設(shè)這4臺(tái)機(jī)器的IP分別為：

10.211.55.4
10.211.55.5
10.211.55.6
10.211.55.7

首先確保幾臺(tái)虛擬機(jī)上安裝了JDK1.8+:

wget --no-check-certificate --no-cookies --header "Cookie: oraclelicense=accept-securebackup-cookie" https://download.oracle.com/otn-pub/java/jdk/8u191-b12/2787e4a523244c269598db4e85c51e0c/jdk-8u191-linux-x64.tar.gz

JAVA_HOME=/opt/java/jdk1.8.0_191
CLASSPATH=$JAVA_HOME/lib/
PATH=$PATH:$JAVA_HOME/bin
export PATH JAVA_HOME CLASSPATH

其次我們打算通過(guò)RocketMQ的源碼進(jìn)行編譯，因?yàn)樵创a是基于Maven開(kāi)發(fā)的Java工程，所以我們需要安裝下Maven環(huán)境:

wget https://archive.apache.org/dist/maven/binaries/apache-maven-3.2.1-bin.tar.gz

export MAVEN_HOME=/opt/apache-maven-3.2.1
export PATH=${PATH}:${MAVEN_HOME}/bin

#配置阿里云鏡像
<mirror>
       <id>nexus-aliyun</id>
       <mirrorOf>central</mirrorOf>
       <name>Nexus aliyun</name>
       <url>http://maven.aliyun.com/nexus/content/groups/public</url>
</mirror>

如果多臺(tái)機(jī)器，沒(méi)有必要依次下載，可以通過(guò)遠(yuǎn)程復(fù)制命令完成機(jī)器間的拷貝:

scp -r /opt/apache-maven-3.2.1/ root@10.211.55.6:/opt/

完成后，我們就可以在主機(jī)的指定目錄下載RocketMQ的源碼發(fā)布包（這里為4.3.2版本）進(jìn)行編譯了：

#download-4.3.2源碼準(zhǔn)備編譯

wget http://mirror.bit.edu.cn/apache/rocketmq/4.3.2/rocketmq-all-4.3.2-source-release.zip 

mvn -Prelease-all -DskipTests clean install -U

cd /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq

以上動(dòng)作需要在各個(gè)機(jī)器節(jié)點(diǎn)也同步操作！編譯完成后，我們來(lái)規(guī)劃下如何利用這4臺(tái)虛擬機(jī)來(lái)實(shí)現(xiàn)“3個(gè)NameSrv節(jié)點(diǎn)、2組Master-Slave Broker集群”的效果。

因?yàn)楸镜貦C(jī)器資源原因，我們通過(guò)虛擬機(jī)混部的方式來(lái)實(shí)現(xiàn)上述集群的效果，4臺(tái)機(jī)器的3臺(tái)會(huì)分別作為NameSrv節(jié)點(diǎn)，而對(duì)于Broker集群則兩兩組合，如上表所示！

按照上述規(guī)劃，接下來(lái)我們就來(lái)看下具體的配置方式：

1）Master-Broker-a的配置
創(chuàng)建數(shù)據(jù)存儲(chǔ)目錄

[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store
[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store/commitlog
[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store/consumequeue
[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store/index
[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store/checkpoint
[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store/abort

切換服務(wù)器目錄為對(duì)應(yīng)的配置文件目錄

cd /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq/conf

cd 2m-2s-async //這里因?yàn)槲覀儾捎玫氖钱惒綇?fù)制模式，所以需要編輯2m-2s-async中的配置文件

編輯Broker集群配置文件
[root@bogon 2m-2s-async]# vim broker-a.properties

#broker所屬哪個(gè)集群，默認(rèn)【DefaultCluster】
brokerClusterName=DefaultCluster

#broker 實(shí)列名稱，主從關(guān)系的需要保持名稱一致
brokerName=broker-a

#brokerId，必須是大等于0的整數(shù)，0表示Master，>0表示Slave
brokerId=0

#刪除文件的時(shí)間點(diǎn)，默認(rèn)為凌晨4點(diǎn)
deleteWhen=04

#文件保留時(shí)間，默認(rèn)為48小時(shí)
fileReservedTime=48

#-ASYNC_MASTER 異步復(fù)制Master
#-SYNC_MASTER  同步雙寫(xiě)Master
#-SLAVE
brokerRole=ASYNC_MASTER

#刷盤(pán)方式
#-ASYNC_FLUSH 異步刷盤(pán)
#-SYNC_FLUSH 同步刷盤(pán)
flushDiskType=ASYNC_FLUSH

#NameSrv集群地址
namesrvAddr=10.211.55.4:9876;10.211.55.5:9876;10.211.55.6:9876

#broker對(duì)外服務(wù)的監(jiān)聽(tīng)端口
listenPort=10911

defaultTopicQueueNums=4

#是否允許broker自動(dòng)創(chuàng)建Topic，建議線下開(kāi)啟，線上關(guān)閉，默認(rèn)【true】
autoCreateTopicEnable=false

#是否允許broker自動(dòng)創(chuàng)建訂閱組，建議線下開(kāi)啟，線上關(guān)閉，默認(rèn)【true】
autoCreateSubscriptionGroup=false
mapedFileSizeCommitLog=1073741824 
mapedFileSizeConsumeQueue=50000000
destroyMapedFileIntervalForcibly=120000
redeleteHangedFileInterval=120000
diskMaxUsedSpaceRatio=88
storePathRootDir=/usr/local/rocketmq/data/store
storePathCommitLog=/usr/local/rocketmq/data/store/commitlog

#消費(fèi)隊(duì)列存儲(chǔ)路徑
storePathConsumeQueue=/usr/local/rocketmq/data/store/consumequeue

#消息索引存儲(chǔ)路徑
storePathIndex=/usr/local/rocketmq/data/store/index

#checkpoint文件存儲(chǔ)路徑
storeCheckpoint=/usr/local/rocketmq/data/store/checkpoint

#abort文件存儲(chǔ)路徑
abortFile=/usr/local/rocketmq/data/store/abort
maxMessageSize=65536
flushCommitLogLeastPages=4
flushConsumeQueueLeastPages=2
flushCommitLogThoroughInterval=10000
flushConsumeQueueThoroughInterval=60000
checkTransactionMessageEnable=false
sendMessageThreadPoolNums=128
pullMessageThreadPoolNums=128

2）、Slave-Broker-a的配置
創(chuàng)建數(shù)據(jù)存儲(chǔ)目錄

[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store
[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store/commitlog
[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store/consumequeue
[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store/index
[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store/checkpoint
[root@bogon local]# mkdir -p /usr/local/rocketmq/data/store/abort

切換服務(wù)器目錄為對(duì)應(yīng)的配置文件目錄

cd /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq/conf

cd 2m-2s-async //這里因?yàn)槲覀儾捎玫氖钱惒綇?fù)制模式，所以需要編輯2m-2s-async中的配置文件

3）、Master-Broker-b的配置

參考Master-Broker-a的配置方式，只需要改下“brokerName=broker-b”即可，其他一樣！

4）、Slave-Broker-b的配置

參考Slave-Broker-a的配置，只需要改下“brokerName=broker-b”即可，其他一樣！

至此，我們就完成了整個(gè)RocketMQ集群的配置了！接下來(lái)我們啟動(dòng)整個(gè)集群。

首先我們需要關(guān)閉各個(gè)節(jié)點(diǎn)的防火墻（很重要）

service iptables stop 
#CentOs7
systemctl stop firewalld

先分別啟動(dòng)三個(gè)NameSrv節(jié)點(diǎn)

cd /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq
nohup sh bin/mqnamesrv &
tail -f ~/logs/rocketmqlogs/namesrv.log

啟動(dòng)Broker集群

#啟動(dòng)Master broker-a服務(wù)（10.211.55.4）
cd /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq/bin/
[root@bogon bin]#
nohup sh mqbroker -c /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq/conf/2m-2s-sync/broker-a.properties >/dev/null 2>&1 &

#啟動(dòng)Slave broker-a-s服務(wù)
cd /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq/bin/
nohup sh mqbroker -c /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq/conf/2m-2s-sync/broker-a-s.properties >/dev/null 2>&1 &

#啟動(dòng)Master broker-b服務(wù)（10.211.55.6）
cd /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq/bin/
nohup sh mqbroker -c /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq/conf/2m-2s-sync/broker-b.properties >/dev/null 2>&1 &

#啟動(dòng)Slave broker-b-s服務(wù)（10.211.55.7）
cd /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq/bin
nohup sh mqbroker -c /opt/rocketmq-all-4.3.2/distribution/target/apache-rocketmq/conf/2m-2s-sync/broker-b-s.properties >/dev/null 2>&1 &

到這里，我們就完成了RocketMQ生產(chǎn)級(jí)集群的模擬搭建，可以通過(guò)jps命令檢查各節(jié)點(diǎn)NameSrv&Broker進(jìn)程是否啟動(dòng)成功。

集群?jiǎn)?dòng)成功后，可以通過(guò)常用命令來(lái)進(jìn)行查看！如：
查看所有消費(fèi)組group

[root@bogon bin]# sh mqadmin consumerProgress -n 10.211.55.6:9876 

查看集群消息

[root@bogon bin]# sh mqadmin clusterList  -n 10.211.55.5:9876