【分佈式事務】基於RocketMQ搭建生產級消息集羣？

導讀

 

目前不少互聯網公司的系統都在朝着微服務化、分佈式化系統的方向在演進，這帶來了不少好處，也帶來了一些棘手的問題，其中最棘手的莫過於數據一致性問題了。早期咱們的軟件功能都在一個進程中，數據的一致性能夠經過數據庫本地事務來加以控制。而在分佈式架構下，本來比較完整的本地功能可能被拆分紅了多個獨立的服務進程。與以前相比，一樣一筆業務訂單此時可能會經歷不少服務模塊的處理，調用鏈路會變得很長，例如某電商平臺，一筆購物訂單可能會通過：商品中心、訂單、支付、物流等多個服務的調用，而這可能還只是比較粗粒度的劃分，某些比較大型的服務，如支付系統，可能自己又會按照分佈式的架構拆分紅多個微服務，因此整個業務的調用鏈路會變得更加冗長。

而這不可避免的就會產生數據不一致的問題，爲了實現業務上的最終一致性，功能比較獨立的系統，如訂單系統與支付系統就會經過額外的業務邏輯設計來確保彼此之間的最終一致性，如訂單系統會經過訂單的支付狀態來保持與支付系統的數據一致，而支付系統則會提供支付狀態查詢接口，或者實現最大可能的主動回調功能，來確保兩者數據狀態的最終一致。此外可能還會經過日終的訂單對帳來發現不一致的數據，並進行數據校訂。

可是這些都只是業務邏輯上的手段，對於某些內部服務之間的調用，若是能夠經過分佈式事務解決方案來加以保證的話，實際上是能夠大大減小一些沒必要要的複雜業務邏輯的。實際上，目前市面上可以提供分佈式事務解決方案、又比較成熟的開源技術框架比較少，而RocketMQ在4.3.0以後的版本提供了事務消息的功能，由於RocketMQ自己擁有比較多的生產實踐的關係，因此這一功能備受關注，做者所在的公司也有一些實踐。

以此爲契機，爲了給你們關於分佈式事務一個比較清晰的認識，這裏我打算以RocketMQ的事務消息功能爲示例，來相對全面的總結下分佈式事務的內容。本篇文章的主要內容，是先介紹如何搭建一套生產級的RocketMQ消息集羣，以此準備下試驗環境。在下一篇《【分佈式事務】基於RocketMQ的分佈式事務實現》會總體介紹下分佈式事務的概念和原理，並作一些代碼級的試驗。

 

什麼是RocketMQ

 

RocketMQ是阿里開源的並貢獻給Apache基金會的一款分佈式消息平臺，它具備低延遲、高性能和可靠性、萬億級容量和靈活的可伸縮性的特色，單機也能夠支持億級的消息堆積能力、單機寫入TPS單實例約7萬條/秒，單機部署3個Broker，能夠跑到最高12萬條/秒。

 

基於以上強大的性能，以及阿里的技術影響力，RocketMQ目前在國內互聯網公司中被使用得比較普遍。那麼，咱們先大概來了解一下RocketMQ的總體結構吧！

 

整個RocketMQ消息系統主要由以下4個部分組成：

 

從中間件服務角度來看整個RocketMQ消息系統（服務端）主要分爲：NameSrv和Broker兩個部分。

 

NameSrv：在RocketMQ分佈式消息系統中，NameSrv主要提供兩個功能：

 

提供服務發現和註冊，這裏主要是管理Broker，NameSrv接受來自Broker的註冊，並經過心跳機制來檢測Broker服務的健康性；

 

提供路由功能，集羣(這裏是指以集羣方式部署的NameSrv)中的每一個NameSrv都保存了Broker集羣(這裏是指以集羣方式部署的Broker)中整個的路由信息和隊列信息。這裏須要注意，在NameSrv集羣中，每一個NameSrv都是相互獨立的，因此每一個Broker須要鏈接全部的NameSrv，每建立一個新的topic都要同步到全部的NameSrv上。

 

Broker：主要是負責消息的存儲、傳遞、查詢以及高可用(HA)保證等。其由以下幾個子模塊（源碼整體上也是這麼拆分的）構成：

 

remoting，是Broker的服務入口，負責客戶端的接入（Producer和Consumer）和請求處理。

 

client，管理客戶端和維護消費者對於Topic的訂閱。

 

store，提供針對存儲和消息查詢的簡單的API（數據存儲在物理磁盤）。

 

HA， 提供數據在主從節點間同步的功能特性。

 

Index，經過特定的key構建消息索引，並提供快速的索引查詢服務。

 

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

 

Producer：消息的生產者，由用戶進行分佈式部署，消息由Producer通過多種負載均衡模式發送到Broker集羣，發送低延時，支持快速失敗。

 

Consumer：消息的消費者，也由用戶部署，支持PUSH和PULL兩種消費模式，支持集羣消費和廣播消息，提供實時的消息訂閱機制，知足大多數消費場景。

 

來總結下，整個RocketMQ消息集羣就是由NameSrv/Broker、Producer/Consumer組成的。爲了讓你們更清晰的理解它們之間的關係，咱們以一條完整的信息流轉爲例，來看看RocketMQ消息系統是如何運轉的，以下圖所示：

 

看到這裏相信你們應該對RocketMQ有一個大體的瞭解了，那麼下面咱們就具體看看，如何搭建一套生產級的RocketMQ消息集羣系統吧！

 

RocketMQ集羣模式

 

RocketMQ集羣部署有多種模式，對於NameSrv來講能夠同時部署多個節點，而且這些節點間也不須要有任何的信息同步，這裏由於每一個NameSrv節點都會存儲全量路由信息，在NameSrv集羣模式下，每一個Broker都須要同時向集羣中的每一個NameSrv節點發送註冊信息，因此這裏對於NameSrv的集羣部署來講並不須要作什麼額外的設置。

 

而對於Broker集羣來講就有多種模式了，這裏我先給你們介紹下這幾種模式，而後咱們再來看看生產級的部署方式是什麼：

 

1）、單個Master模式

 

一個Broker做爲主服務，不設置任何Slave，很顯然這種方式風險比較大，存在單節點故障會致使整個基於消息的服務掛掉，因此生產環境不可能採用這種模式。

 

2）、多Master模式

 

這種模式的Broker集羣，全是Master，沒有Slave節點。這種方式的優缺點以下：

 

優勢：配置會比較簡單一些，若是單個Master掛掉或重啓維護的話對應用是沒有什麼影響的。若是磁盤配置爲RAID10（服務器的磁盤陣列模式，遺忘的同窗能夠本身查下資料）的話，即便在機器宕機不可恢復的狀況下，因爲RAID10磁盤自己的可靠性，消息也不會丟失（異步刷盤丟失少許消息，同步刷盤一條不丟），這種Broker的集羣模式性能相對來講是最高的。

 

缺點：就是在單臺機器宕機期間，這臺機器上未被消費的消息在機器恢復以前是不能夠進行消息訂閱的，這對消息的實時性會有一些影響。

 

3）、多Master多Slave模式（異步複製）

 

在這種模式下Broker集羣存在多個Master節點，而且每一個Master節點都會對應一個Slave節點，有多對Master-Slave，HA(高可用)之間採用異步複製的方式進行信息同步，在這種方式下主從之間會有短暫的毫秒級的消息延遲。

 

優勢：在這種模式下即使磁盤損壞了，消息丟失的狀況也很是少，由於主從之間有信息備份；而且，在這種模式下消息的實時性也不會受影響，由於Master宕機後Slave能夠自動切換爲Master模式，這樣Consumer仍然能夠經過Slave進行消息消費，而這個過程對應用來講則是徹底透明的，並不須要人工干預；另外，這種模式的性能與多Master模式幾乎差很少。

 

缺點：若是Master宕機，而且在磁盤損壞的狀況下，會丟失少許的消息。

 

4）、多Master多Slave模式（同步複製）

 

這種模式與3）差很少，只是HA採用的是同步雙寫的方式，即主備都寫成功後，纔會嚮應用返回成功。

 

優勢：在這種模式下數據與服務都不存在單點的狀況，在Master宕機的狀況下，消息也沒有延遲，服務的可用性以及數據的可用性都很是高。

 

缺點：性能相比於異步複製來講略低一些（大約10%）；另一個缺點就是相比於異步複製，目前Slave備機還暫時不能實現自動切換爲Master，可能後續的版本會支持Master-Slave的自動切換功能。

 

生產級RocketMQ集羣

 

綜合考慮以上集羣模式的優缺點，在實際生產環境中目前基於RocketMQ消息集羣的部署方式基本都是採用多Master多Slave（異步複製）這種模式，做者目前所在公司的生產環境的Rocket消息系統也是採用這種模式進行部署的。

 

如下爲目前做者所在公司的實際部署結構：

 

在以上實踐中，部署了3個NameSrv節點，Broker採用2主2從的異步複製模式進行集羣部署。

 

爲了更好地理解RocketMQ的集羣運行原理，接下來咱們以4臺虛擬機來模擬上述集羣的搭建過程，假設這4臺機器的IP分別爲：

 

10.211.55.4
10.211.55.5
10.211.55.6

10.211.55.7

 

首先確保幾臺虛擬機上安裝了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

 

其次咱們打算經過RocketMQ的源碼進行編譯，由於源碼是基於Maven開發的Java工程，因此咱們須要安裝下Maven環境:

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>

 

若是多臺機器，沒有必要依次下載，能夠經過遠程複製命令完成機器間的拷貝:

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

 

完成後，咱們就能夠在主機的指定目錄下載RocketMQ的源碼發佈包（這裏爲4.3.2版本）進行編譯了：

#download-4.3.2源碼準備編譯

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

 

以上動做須要在各個機器節點也同步操做！編譯完成後，咱們來規劃下如何利用這4臺虛擬機來實現「3個NameSrv節點、2組Master-Slave Broker集羣」的效果。

 

1）Master-Broker-a的配置

 

#建立數據存儲目錄

[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

 

#切換服務器目錄爲對應的配置文件目錄

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

cd 2m-2s-async //這裏由於咱們採用的是異步複製模式，因此須要編輯2m-2s-async中的配置文件

 

#編輯Broker集羣配置文件

[root@bogon 2m-2s-async]# vim broker-a.properties

#broker所屬哪一個集羣，默認【DefaultCluster】
brokerClusterName=DefaultCluster

#broker 實列名稱，主從關係的須要保持名稱一致
brokerName=broker-a

#brokerId，必須是大等於0的整數，0表示Master，>0表示Slave
brokerId=0

#刪除文件的時間點，默認爲凌晨4點
deleteWhen=04

#文件保留時間，默認爲48小時
fileReservedTime=48

#-ASYNC_MASTER 異步複製Master
#-SYNC_MASTER  同步雙寫Master
#-SLAVE
brokerRole=ASYNC_MASTER

#刷盤方式
#-ASYNC_FLUSH 異步刷盤
#-SYNC_FLUSH 同步刷盤
flushDiskType=ASYNC_FLUSH

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

#broker對外服務的監聽端口
listenPort=10911

defaultTopicQueueNums=4

#是否容許broker自動建立Topic，建議線下開啓，線上關閉，默認【true】
autoCreateTopicEnable=false

#是否容許broker自動建立訂閱組，建議線下開啓，線上關閉，默認【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

#消費隊列存儲路徑
storePathConsumeQueue=/usr/local/rocketmq/data/store/consumequeue

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

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

#abort文件存儲路徑
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的配置

#建立數據存儲目錄

[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

 

#切換服務器目錄爲對應的配置文件目錄

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

cd 2m-2s-async //這裏由於咱們採用的是異步複製模式，因此須要編輯2m-2s-async中的配置文件

 

#修改Slaves

[root@localhost 2m-2s-async]# vim broker-a-s.properties

 

#broker所屬的那個集羣，默認【DefaultCluster】
brokerClusterName=DefaultCluster

#broker實例名稱，主從關係的須要保持名稱一致
brokerName=broker-a

#brokerid，必須是大於0的整數，0表示Master，>0表示Slave，
#一個Master能夠掛多個Slave，Master與Slave經過brokerName來配對，默認【0】
brokerId=1

#刪除文件時間點，默認爲凌晨4點
deleteWhen=04

#文件保留時間，默認48小時
fileReservedTime=48

#broker角色，這裏爲Slave節點
brokerRole=SLAVE

#刷盤方式 -ASYNC_FLUSH 異步刷盤；-SYNC_FLUSH 同步刷盤
flushDiskType=ASYNC_FLUSH

#NameSrv節點地址
namesrvAddr=10.211.55.4:9876;10.211.55.5:9876;10.211.55.6:9876

#Broker對外服務的監聽端口，默認【10911】
listenPort=10911
defaultTopicQueueNums=4 

#是否容許broker自動建立Topic，建議線下開啓，線上關閉，默認【true】
autoCreateTopicEnable=false

#是否容許broker自動建立訂閱組，建議線下開啓，線上關閉，默認【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

#消費隊列存儲路徑
storePathConsumeQueue=/usr/local/rocketmq/data/store/consumequeue

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

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

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

 

3）、Master-Broker-b的配置

 

參考Master-Broker-a的配置方式，只須要改下「brokerName=broker-b」便可，其餘同樣！

 

4）、Slave-Broker-b的配置

 

參考Slave-Broker-a的配置，只須要改下「brokerName=broker-b」便可，其餘同樣！

 

至此，咱們就完成了整個RocketMQ集羣的配置了！接下來咱們啓動整個集羣。

 

#首先咱們須要關閉各個節點的防火牆（很重要）

service iptables stop 
#CentOs7
systemctl stop firewalld

 

#先分別啓動三個NameSrv節點

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

 

#啓動Broker集羣

#啓動Master broker-a服務（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 &

#啓動Slave broker-a-s服務
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 &

#啓動Master broker-b服務（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 &

#啓動Slave broker-b-s服務（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生產級集羣的模擬搭建，能夠經過jps命令檢查各節點NameSrv&Broker進程是否啓動成功。

 

集羣啓動成功後，能夠經過經常使用命令來進行查看！如：

#查看全部消費組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

 

 因爲篇幅的關係到這裏本篇文章的內容就結束了！在下一篇文章中咱們會詳細介紹分佈式事務的概念，敬請關注！

備註：本文首發於微信公衆號「無敵碼農」，感興趣的朋友也能夠關注下微信公衆號，提早收看更多精彩內容！