RocketMQ安裝部署

• 一、單機(jī)部署

• 1.4.1、下載

• 1.4.2、修改配置

• 1.4.3、添加依賴

• 1.4.4、打包

• 1.4.5、啟動(dòng)

• 1.4.6、訪問

• 1.3.1、啟動(dòng) NameServer

• 1.3.2、啟動(dòng) Broker

• 1.3.3、發(fā)送/接收消息測試

• 1.3.4、關(guān)閉服務(wù)

• 1.2.1、runserver.sh

• 1.2.2、runbroker.sh

• 1.2.3、tools.sh

• 1.2.4、自動(dòng)動(dòng)創(chuàng)建Topic功能

• 1.1、下載

• 1.2、修改RocketMQ啟動(dòng)配置

• 1.3、RocketMQ 啟動(dòng)

• 1.4、控制臺(tái)的安裝與啟動(dòng)

• 二、集群搭建理論

• 2.1.1、復(fù)制策略

• 2.1.2、刷盤策略

• 2.1、數(shù)據(jù)復(fù)制與刷盤策略

• 2.2、Broker集群模式

• 三、搭建主從異步集群

• 3.3.1、broker-a的master節(jié)點(diǎn)

• 3.3.2、broker-a的slave節(jié)點(diǎn)

• 3.3.3、broker-b的master節(jié)點(diǎn)

• 3.3.4、broker-b的slave節(jié)點(diǎn)

• 3.3.5、啟動(dòng)broker

• 3.1、集群架構(gòu)

• 3.2、啟動(dòng)三臺(tái)nameserver

• 3.3、配置broker

• 3.4、驗(yàn)證集群

• 四、mqadmin管理工具

• 4.1、創(chuàng)建topic：updateTopic

• 4.2、刪除Topic：deleteTopic

• 4.3、創(chuàng)建（修訂）訂閱組：updateSubGroup

• 4.4、刪除訂閱組配置：deleteSubGroup

• 4.5、更新Broker 配置?件：updateBrokerConfig

• 4.6、查看Topic 列表信息：topicList

• 4.7、查看Topic 路由信息：topicRoute

• 4.8、查看Topic 統(tǒng)計(jì)信息：topicStats

• 4.9、查看Broker 統(tǒng)計(jì)信息：brokerStats

• 4.10、根據(jù)消息ID 查詢消息：queryMsgById

• 4.11、根據(jù)消息Key 查詢消息：queryMsgByKey

• 4.12、根據(jù)Offset 查詢消息：queryMsgByOffset

• 4.13、查詢Producer 的網(wǎng)絡(luò)連接：producerConnection

• 4.14、查詢Consumer 的?絡(luò)連接：consumerConnection

• 4.15、查看訂閱組消費(fèi)狀態(tài)：consumerProgress

• 4.16、查看集群消息：clusterList

• 4.18、添加（更新）KV 配置信息：updateKvConfig

• 4.19、刪除KV 配置信息：deleteKvConfig

• 4.20、添加（更新）Project group 配置信息：updateProjectGroup

• 4.21、刪除Project group 配置信息：deleteProjectGroup

• 4.22、取得Project group 配置信息：getProjectGroup

• 4.23、設(shè)置消費(fèi)進(jìn)度：resetOffsetByTime

• 4.24、清除特定Broker權(quán)限：wipeWritePerm

• 4.25、獲取Consumer消費(fèi)進(jìn)度：getConsumerStatus

• 五、安裝可視化管理控制平臺(tái)

一、單機(jī)部署

環(huán)境說明

對于 RocketMQ 4.3.0版本，官方要求環(huán)境如下，其中 Git 用于從 GitHub 獲取源碼，沒有安裝也沒關(guān)系，可以直接下載
官方推薦的流程是：Linux 系統(tǒng)上安裝 Git 工具 、Maven、Java JDK
Git 工具用于直接從 GitHub 獲取 RocketMQ 項(xiàng)目源碼下載到 Linux 系統(tǒng)上
然后 Maven 將 RocketMQ 源碼進(jìn)行編譯成二進(jìn)制文件
安裝了 Java JDK 就可以運(yùn)行 RocketMQ 了

1.1、下載

## 切換/opt/home目錄下cd /opt/home## 下載wget https://archive.apache.org/dist/rocketmq/4.8.0/rocketmq-all-4.8.0-bin-release.zip## 項(xiàng)目解壓unzip rocketmq-all-4.8.0-bin-release.zip## 重命名mv rocketmq-all-4.8.0-bin-release rocketmq4.8.01234567891011

1.2、修改RocketMQ啟動(dòng)配置

修改目錄 /opt/home/rocketmq4.8.0/bin 下的 3 個(gè)配置文件： runserver.sh、runbroker.sh 、tools.sh不然會(huì)報(bào)insufficient memory

cd /opt/home/rocketmq4.8.0/bin1

1.2.1、runserver.sh

vi runserver.sh# JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn512m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"1234

1.2.2、runbroker.sh

vi runbroker.sh# JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"1234

1.2.3、tools.sh

vi tools.sh# JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=128m"JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=128m"1234

1.2.4、自動(dòng)動(dòng)創(chuàng)建Topic功能

在conf/broker.conf?件中加?如下配置，開啟自動(dòng)動(dòng)創(chuàng)建Topic功能，不建議開啟。

autoCreateTopicEnable=true1

1.3、RocketMQ 啟動(dòng)

1.3.1、啟動(dòng) NameServer

進(jìn)入 RocketMQ 安裝目錄下的/opt/home/rocketmq4.8.0/bin目錄進(jìn)行操作

cd /opt/home/rocketmq4.8.0/bin1

執(zhí)行命令啟動(dòng)NameServer

## 創(chuàng)建日志目錄 mkdir logs# nohup ./mqnamesrv &：屬于后臺(tái)以靜默?式啟動(dòng)# ./mqnamesrv：屬于終端啟動(dòng)，直接輸出日志信息，按 ctrl+c 可直接關(guān)閉退出nohup ./mqnamesrv > logs/mqnamesrv.out 2>1 &1234567

查看啟動(dòng)狀態(tài)，在當(dāng)前目錄下會(huì)有一個(gè)nohup.out的日志文件，可以打開查看。

## 查看日志tail -f logs/mqnamesrv.out## 看到以下表示啟動(dòng)成功The Name Server boot success. serializeType=JSON12345

解決報(bào)錯(cuò)

## 報(bào)錯(cuò)ERROR: Please set the JAVA_HOME variable in your environment, We need java(x64)! !!## 解決 配置jdk環(huán)境變量# 需要 export 環(huán)境變量12345

1.3.2、啟動(dòng) Broker

同樣進(jìn)入 RocketMQ 安裝目錄下的 /opt/home/rocketmq4.8.0/bin目錄進(jìn)行操作

執(zhí)行啟動(dòng)命令，并且常駐內(nèi)存，注意ip地址要配置成為服務(wù)的ip地址，保證地址以及端口能夠訪問。

# 啟動(dòng)命令，并且常駐內(nèi)存:注意ip地址要配置成為服務(wù)的ip地址，保證地址以及端口能夠訪問# nohup ./mqbroker -n 192.168.0.101:9876 & ：屬于后臺(tái)以靜默?式啟動(dòng)# sh ./mqbroker -n 92.168.0.101:9876 ：屬于終端啟動(dòng)，直接輸出日志信息，按 ctrl+c 可直接關(guān)閉退出nohup ./mqbroker -n 192.168.0.101:9876 > logs/mqbroker.out 2>1 &12345

查看啟動(dòng)狀態(tài)，啟動(dòng)之后同樣提示將日志信息追加到了當(dāng)前目錄下的nohup.out文件中。

## 查看日志tail -f logs/mqbroker.out## 看到以下表示啟動(dòng)成功The broker[linux1, 192.168.0.101:10911] boot success. serializeType=JSON and name server is 192.168.0.101:987612345

1.3.3、發(fā)送/接收消息測試

發(fā)送/接收消息之前，需要告訴客戶端(Producer、Consumer)名稱服務(wù)器的位置，RocketMQ 提供了多種方法來實(shí)現(xiàn)這一點(diǎn)

• 編程方式，如：producer.setNamesrvAddr("ip:port")

• Java 選項(xiàng)，如：rocketmq.namesrv.addr

• 環(huán)境變量，如：NAMESRV_ADDR

• HTTP 端點(diǎn)

發(fā)送消息

export NAMESRV_ADDR=192.168.0.101:9876
./tools.sh org.apache.rocketmq.example.quickstart.Producer12

示例：

[root@linux1 rocketmq]# export NAMESRV_ADDR=localhost:9876[root@linux1 rocketmq]# sh bin/tools.sh org.apache.rocketmq.example.quickstart.ProducerOpenJDK 64-Bit Server VM warning: MaxNewSize (262144k) is equal to or greater than the entire heap (262144k).  A new max generation size of 261632k will be used.16:19:05.806 [main] DEBUG i.n.u.i.l.InternalLoggerFactory - Using SLF4J as the default logging framework
RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.PlatformDependent0).
RocketMQLog:WARN Please initialize the logger system properly.
SendResult [sendStatus=SEND_OK, msgId=AC11000176396FF3C5B512F379FA0000, offsetMsgId=AC11000100002A9F0000000000000000, messageQueue=MessageQueue [topic=TopicTest, brokerName=linux1, queueId=3], queueOffset=0]......SendResult [sendStatus=SEND_OK, msgId=AC11000176396FF3C5B512F382B603E7, offsetMsgId=AC11000100002A9F00000000000317BF, messageQueue=MessageQueue [topic=TopicTest, brokerName=linux1, queueId=2], queueOffset=249]16:19:08.609 [NettyClientSelector_1] INFO  RocketmqRemoting - closeChannel: close the connection to remote address[172.17.0.1:10911] result: true16:19:08.631 [NettyClientSelector_1] INFO  RocketmqRemoting - closeChannel: close the connection to remote address[127.0.0.1:9876] result: true1234567891011

接收消息

./tools.sh org.apache.rocketmq.example.quickstart.Consumer1

示例

[root@linux1 rocketmq]# sh bin/tools.sh org.apache.rocketmq.example.quickstart.ConsumerOpenJDK 64-Bit Server VM warning: MaxNewSize (262144k) is equal to or greater than the entire heap (262144k).  A new max generation size of 261632k will be used.16:21:15.395 [main] DEBUG i.n.u.i.l.InternalLoggerFactory - Using SLF4J as the default logging framework
Consumer Started.
ConsumeMessageThread_3 Receive New Messages: [MessageExt [brokerName=linux1, queueId=2, storeSize=201, queueOffset=1, sysFlag=0, bornTimestamp=1659601146477, bornHost=/192.168.0.101:48216, storeTimestamp=1659601146478, storeHost=/172.17.0.1:10911, msgId=AC11000100002A9F000000000000057F, commitLogOffset=1407, bodyCRC=988340972, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=250, CONSUME_START_TIME=1659601275866, UNIQ_KEY=AC11000176396FF3C5B512F37A6D0007, CLUSTER=DefaultCluster, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 55], transactionId='null'}]] ConsumeMessageThread_4 Receive New Messages: [MessageExt [brokerName=linux1, queueId=2, storeSize=202, queueOffset=2, sysFlag=0, bornTimestamp=1659601146500, bornHost=/192.168.0.101:48216, storeTimestamp=1659601146501, storeHost=/172.17.0.1:10911, msgId=AC11000100002A9F00000000000008A4, commitLogOffset=2212, bodyCRC=2088767104, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=250, CONSUME_START_TIME=1659601275867, UNIQ_KEY=AC11000176396FF3C5B512F37A84000B, CLUSTER=DefaultCluster, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 49, 49], transactionId='null'}]]123456

消息發(fā)送完畢之后就會(huì)退出，在同一窗口中可以使用消費(fèi)者類來進(jìn)行接收消息，消費(fèi)是多線程的。

1.3.4、關(guān)閉服務(wù)

與啟動(dòng)順序相反進(jìn)行關(guān)閉，先關(guān)閉 broker、在關(guān)閉 nameserv

./mqshutdown broker
./mqshutdown namesrv12

1.4、控制臺(tái)的安裝與啟動(dòng)

RocketMQ有一個(gè)可視化的dashboard，通過該控制臺(tái)可以直觀的查看到很多數(shù)據(jù)。

1.4.1、下載

下載地址：https://github.com/apache/rocketmq-externals/tags

下載rocketmq-console-1.0.0：https://github.com/apache/rocketmq-externals/archive/refs/tags/rocketmq-console-1.0.0.zip

1.4.2、修改配置

使用IDEA打開項(xiàng)目，修改其src/main/resources中的application.properties配置文件。

• 原來的端口號為8080，如果8080端口被占用，修改為一個(gè)不常用的端口。

• 指定RocketMQ的name server地址，集群環(huán)境為集群1:9876;集群2:9876;集群3:9876。

server.contextPath=
server.port=7000
#spring.application.index=true
spring.application.name=rocketmq-console
spring.http.encoding.charset=UTF-8
spring.http.encoding.enabled=true
spring.http.encoding.force=true
logging.config=classpath:logback.xml
#if this value is empty,use env value rocketmq.config.namesrvAddr  NAMESRV_ADDR | now, you can set it in ops page.default localhost:9876
rocketmq.config.namesrvAddr=localhost:9876
#if you use rocketmq version < 3.5.8, rocketmq.config.isVIPChannel should be false.default true
rocketmq.config.isVIPChannel=
#rocketmq-console's data path:dashboard/monitor
rocketmq.config.dataPath=/tmp/rocketmq-console/data
#set it false if you don't want use dashboard.default true
rocketmq.config.enableDashBoardCollect=true12345678910111213141516

1.4.3、添加依賴

在解壓目錄rocketmq-console的pom.xml中添加如下JAXB依賴。

JAXB，Java Architechture for Xml Binding，用于XML綁定的Java技術(shù)，是一個(gè)業(yè)界標(biāo)準(zhǔn)，是一 項(xiàng)可以根據(jù)XML Schema生成Java類的技術(shù)。

    javax.xml.bind    jaxb-api    2.3.0    com.sun.xml.bind    jaxb-impl    2.3.0    com.sun.xml.bind    jaxb-core    2.3.0    javax.activation    activation    1.1.11234567891011121314151617181920

1.4.4、打包

在rocketmq-console目錄下，打開cmd窗口運(yùn)行maven的打包命令。

mvn clean package -Dmaven.test.skip=true1

1.4.5、啟動(dòng)

這里為了方便，我將jar包上傳到服務(wù)器，然后啟動(dòng)。

nohup java -jar rocketmq-console-ng-1.0.0.jar > rocketmq-console.out 2>1 &1

1.4.6、訪問

訪問地址：http://192.168.0.101:8080/#/

更換語言為中文

二、集群搭建理論

2.1、數(shù)據(jù)復(fù)制與刷盤策略

2.1.1、復(fù)制策略

復(fù)制策略是Broker的Master與Slave間的數(shù)據(jù)同步方式。分為同步復(fù)制與異步復(fù)制：

• 同步復(fù)制：消息寫入master后，master會(huì)等待slave同步數(shù)據(jù)成功后才向producer返回成功ACK。

• 異步復(fù)制：消息寫入master后，master立即向producer返回成功ACK，無需等待slave同步數(shù)據(jù)成功。

異步復(fù)制策略會(huì)降低系統(tǒng)的寫入延遲，RT變小，提高了系統(tǒng)的吞吐量。

2.1.2、刷盤策略

刷盤策略指的是broker中消息的落盤方式，即消息發(fā)送到broker內(nèi)存后消息持久化到磁盤的方式。分為 同步刷盤與異步刷盤：

• 同步刷盤：當(dāng)消息持久化到broker的磁盤后才算是消息寫入成功。

• 異步刷盤：當(dāng)消息寫入到broker的內(nèi)存后即表示消息寫入成功，無需等待消息持久化到磁盤。

1.異步刷盤策略會(huì)降低系統(tǒng)的寫入延遲，RT變小，提高了系統(tǒng)的吞吐。
2.消息寫入到Broker的內(nèi)存，一般是寫入到了PageCache。
3.對于異步 刷盤策略，消息會(huì)寫入到PageCache后立即返回成功ACK。但并不會(huì)立即做落盤操 作，而是當(dāng)PageCache到達(dá)一定量時(shí)會(huì)自動(dòng)進(jìn)行落盤。

2.2、Broker集群模式

為了追求更好的性能，RocketMQ的最佳實(shí)踐方式都是在集群模式下完成。RocketMQ官方提供了三種集群搭建方式。

• 2主2從異步通信方式
  使用異步方式進(jìn)行主從之間的數(shù)據(jù)復(fù)制，吞吐量大，但可能會(huì)丟消息。
  使用conf/2m-2s-async文件夾內(nèi)的配置?件做集群配置。

• 2主2從同步通信方式
  使用同步方式進(jìn)行主從之間的數(shù)據(jù)復(fù)制，保證消息安全投遞，不會(huì)丟失，但影響吞吐量
  使用conf/2m-2s-sync文件夾內(nèi)的配置文件做集群配置。

• 2主無從方式
  會(huì)存在單點(diǎn)故障，且讀的性能沒有前兩種方式好。
  使用conf/2m-noslave文件夾內(nèi)的配置文件做集群配置。

• Dledger高可用集群
  上述三種官方提供的集群沒辦法實(shí)現(xiàn)高可用，即在master節(jié)點(diǎn)掛掉后，slave節(jié)點(diǎn)沒辦法自動(dòng)被選舉為新的master，而需要人工實(shí)現(xiàn)。

RocketMQ在4.5版本之后引?了第三方的Dleger?可用集群。

三、搭建主從異步集群

3.1、集群架構(gòu)

這里要搭建一個(gè)雙主雙從異步復(fù)制的Broker集群。為了方便，這里使用了兩臺(tái)主機(jī)來完成集群的搭建。 這兩臺(tái)主機(jī)的功能與broker角色分配如下表。

三臺(tái)服務(wù)器都需要安裝JDK和RocketMQ，安裝步驟參考如上。

3.2、啟動(dòng)三臺(tái)nameserver

nameserver是?個(gè)輕量級的注冊中心，broker把自己的信息注冊到nameserver上。而且，nameserver是無狀態(tài)的，直接啟動(dòng)即可。三臺(tái)nameserver之間不需要通信，而是被請求方來關(guān)聯(lián)三臺(tái)nameserver的地址。

在每臺(tái)服務(wù)器的bin目錄下執(zhí)行如下命令：

## 切換到bin目錄cd /opt/home/rocketmq4.8.0/bin## 創(chuàng)建日志目錄mkdir logs12345

服務(wù)器1

## 后臺(tái)靜默啟動(dòng)nohup ./mqnamesrv -n 192.168.0.101:9876 > logs/mqnamesrv.out 2>1 &12

服務(wù)器2

## 后臺(tái)靜默啟動(dòng)nohup ./mqnamesrv -n 192.168.0.102:9876 > logs/mqnamesrv.out 2>1 &12

服務(wù)器3

## 后臺(tái)靜默啟動(dòng)nohup ./mqnamesrv -n 192.168.0.103:9876 > logs/mqnamesrv.out 2>1 &12

3.3、配置broker

broker-a、broker-b-s這兩臺(tái)broker是配置在服務(wù)器1上，broker-b、broker-a-s這兩臺(tái)broker是配置在服務(wù)器2上。這兩對主從節(jié)點(diǎn)在不同的服務(wù)器上，服務(wù)器3上沒有部署broker。

需要修改每臺(tái)broker的配置文件。注意，同?臺(tái)服務(wù)器上的兩個(gè)broker保存路徑不能?樣。

3.3.1、broker-a的master節(jié)點(diǎn)

在服務(wù)器1上，進(jìn)?到conf/2m-2s-async?件夾內(nèi)，修改broker-a.properties?件。

cd /opt/home/rocketmq4.8.0/conf/2m-2s-asyncvi broker-a.properties# 指定整個(gè)broker集群的名稱，或者說是RocketMQ集群的名稱brokerClusterName=DefaultCluster# 指定master-slave集群的名稱。一個(gè)RocketMQ集群可以包含多個(gè)master-slave集群brokerName=broker-a# broker所在服務(wù)器的ipbrokerIP1=192.168.0.101# broker的id，0表示master，>0表示slavebrokerId=0# 指定刪除消息存儲(chǔ)過期文件的時(shí)間為凌晨4點(diǎn)deleteWhen=04# 指定未發(fā)生更新的消息存儲(chǔ)文件的保留時(shí)長為48小時(shí)，48小時(shí)后過期，將會(huì)被刪除fileReservedTime=48# 指定當(dāng)前broker為異步復(fù)制masterbrokerRole=ASYNC_MASTER# 指定刷盤策略為異步刷盤flushDiskType=ASYNC_FLUSH# 指定Name Server的地址namesrvAddr=192.168.0.101:9876;192.168.0.102:9876;192.168.0.103:9876# 在發(fā)送消息?動(dòng)創(chuàng)建不存在的topic時(shí)，默認(rèn)創(chuàng)建的隊(duì)列數(shù)為4個(gè)defaultTopicQueueNums=4# 是否允許 Broker 自動(dòng)創(chuàng)建Topic，建議線下開啟，線上關(guān)閉autoCreateTopicEnable=true# 是否允許 Broker 自動(dòng)創(chuàng)建訂閱組，建議線下開啟，線上關(guān)閉autoCreateSubscriptionGroup=true# broker對外服務(wù)的監(jiān)聽端口listenPort=10911# abort文件存儲(chǔ)路徑abortFile=/data/rocketmq/store/abort# 消息存儲(chǔ)路徑storePathRootDir=/data/rocketmq/store# commitLog存儲(chǔ)路徑storePathCommitLog=/data/rocketmq/store/commitlog# 消費(fèi)隊(duì)列存儲(chǔ)路徑storePathConsumeQueue=/data/rocketmq/store/consumequeue# 消息索引存儲(chǔ)路徑storePathIndex=/data/rocketmq/store/index# checkpoint文件存儲(chǔ)路徑storeCheckpoint=/data/rocketmq/store/checkpoint# 限制的消息大小maxMessageSize=655361234567891011121314151617181920212223242526272829303132333435363738394041424344

3.3.2、broker-a的slave節(jié)點(diǎn)

在服務(wù)器2上，進(jìn)?到conf/2m-2s-async文件夾內(nèi)，修改broker-a-s.properties文件。

cd /opt/home/rocketmq4.8.0/conf/2m-2s-asyncvi broker-a-s.properties# 指定整個(gè)broker集群的名稱，或者說是RocketMQ集群的名稱brokerClusterName=DefaultCluster# 指定master-slave集群的名稱。一個(gè)RocketMQ集群可以包含多個(gè)master-slave集群brokerName=broker-a# broker所在服務(wù)器的ipbrokerIP1=192.168.0.102# broker的id，0表示master，>0表示slavebrokerId=1# 指定刪除消息存儲(chǔ)過期文件的時(shí)間為凌晨4點(diǎn)deleteWhen=04# 指定未發(fā)生更新的消息存儲(chǔ)文件的保留時(shí)長為48小時(shí)，48小時(shí)后過期，將會(huì)被刪除fileReservedTime=48# 指定當(dāng)前broker為異步復(fù)制masterbrokerRole=SLAVE# 指定刷盤策略為異步刷盤flushDiskType=ASYNC_FLUSH# 指定Name Server的地址namesrvAddr=192.168.0.101:9876;192.168.0.102:9876;192.168.0.103:9876# 在發(fā)送消息?動(dòng)創(chuàng)建不存在的topic時(shí)，默認(rèn)創(chuàng)建的隊(duì)列數(shù)為4個(gè)defaultTopicQueueNums=4# 是否允許 Broker 自動(dòng)創(chuàng)建Topic，建議線下開啟，線上關(guān)閉autoCreateTopicEnable=true# 是否允許 Broker 自動(dòng)創(chuàng)建訂閱組，建議線下開啟，線上關(guān)閉autoCreateSubscriptionGroup=true# broker對外服務(wù)的監(jiān)聽端口listenPort==11011# abort文件存儲(chǔ)路徑abortFile=/data/rocketmq/store-slave/abort# 消息存儲(chǔ)路徑storePathRootDir=/data/rocketmq/store-slave# commitLog存儲(chǔ)路徑storePathCommitLog=/data/rocketmq/store-slave/commitlog# 消費(fèi)隊(duì)列存儲(chǔ)路徑storePathConsumeQueue=/data/rocketmq/store-slave/consumequeue# 消息索引存儲(chǔ)路徑storePathIndex=/data/rocketmq/store-slave/index# checkpoint文件存儲(chǔ)路徑storeCheckpoint=/data/rocketmq/store-slave/checkpoint# 限制的消息大小maxMessageSize=655361234567891011121314151617181920212223242526272829303132333435363738394041424344

3.3.3、broker-b的master節(jié)點(diǎn)

在服務(wù)器2上，進(jìn)入到conf/2m-2s-async文件夾內(nèi)，修改broker-b.properties文件。

cd /opt/home/rocketmq4.8.0/conf/2m-2s-asyncvi broker-b.properties# 指定整個(gè)broker集群的名稱，或者說是RocketMQ集群的名稱brokerClusterName=DefaultCluster# 指定master-slave集群的名稱。一個(gè)RocketMQ集群可以包含多個(gè)master-slave集群brokerName=broker-b# broker所在服務(wù)器的ipbrokerIP1=192.168.0.102# broker的id，0表示master，>0表示slavebrokerId=0# 指定刪除消息存儲(chǔ)過期文件的時(shí)間為凌晨4點(diǎn)deleteWhen=04# 指定未發(fā)生更新的消息存儲(chǔ)文件的保留時(shí)長為48小時(shí)，48小時(shí)后過期，將會(huì)被刪除fileReservedTime=48# 指定當(dāng)前broker為異步復(fù)制masterbrokerRole=ASYNC_MASTER# 指定刷盤策略為異步刷盤flushDiskType=ASYNC_FLUSH# 指定Name Server的地址namesrvAddr=192.168.0.101:9876;192.168.0.102:9876;192.168.0.103:9876# 在發(fā)送消息?動(dòng)創(chuàng)建不存在的topic時(shí)，默認(rèn)創(chuàng)建的隊(duì)列數(shù)為4個(gè)defaultTopicQueueNums=4# 是否允許 Broker 自動(dòng)創(chuàng)建Topic，建議線下開啟，線上關(guān)閉autoCreateTopicEnable=true# 是否允許 Broker 自動(dòng)創(chuàng)建訂閱組，建議線下開啟，線上關(guān)閉autoCreateSubscriptionGroup=true# broker對外服務(wù)的監(jiān)聽端口listenPort=10911# abort文件存儲(chǔ)路徑abortFile=/data/rocketmq/store/abort# 消息存儲(chǔ)路徑storePathRootDir=/data/rocketmq/store# commitLog存儲(chǔ)路徑storePathCommitLog=/data/rocketmq/store/commitlog# 消費(fèi)隊(duì)列存儲(chǔ)路徑storePathConsumeQueue=/data/rocketmq/store/consumequeue# 消息索引存儲(chǔ)路徑storePathIndex=/data/rocketmq/store/index# checkpoint文件存儲(chǔ)路徑storeCheckpoint=/data/rocketmq/store/checkpoint# 限制的消息大小maxMessageSize=655361234567891011121314151617181920212223242526272829303132333435363738394041424344

3.3.4、broker-b的slave節(jié)點(diǎn)

在服務(wù)器1上，進(jìn)入到conf/2m-2s-async文件夾內(nèi)，修改broker-b-s.properties文件。

cd /opt/home/rocketmq4.8.0/conf/2m-2s-asyncvi broker-b-s.properties# 指定整個(gè)broker集群的名稱，或者說是RocketMQ集群的名稱brokerClusterName=DefaultCluster# 指定master-slave集群的名稱。一個(gè)RocketMQ集群可以包含多個(gè)master-slave集群brokerName=broker-b# broker所在服務(wù)器的ipbrokerIP1=192.168.0.101# broker的id，0表示master，>0表示slavebrokerId=1# 指定刪除消息存儲(chǔ)過期文件的時(shí)間為凌晨4點(diǎn)deleteWhen=04# 指定未發(fā)生更新的消息存儲(chǔ)文件的保留時(shí)長為48小時(shí)，48小時(shí)后過期，將會(huì)被刪除fileReservedTime=48# 指定當(dāng)前broker為異步復(fù)制masterbrokerRole=SLAVE# 指定刷盤策略為異步刷盤flushDiskType=ASYNC_FLUSH# 指定Name Server的地址namesrvAddr=192.168.0.101:9876;192.168.0.102:9876;192.168.0.103:9876# 在發(fā)送消息?動(dòng)創(chuàng)建不存在的topic時(shí)，默認(rèn)創(chuàng)建的隊(duì)列數(shù)為4個(gè)defaultTopicQueueNums=4# 是否允許 Broker 自動(dòng)創(chuàng)建Topic，建議線下開啟，線上關(guān)閉autoCreateTopicEnable=true# 是否允許 Broker 自動(dòng)創(chuàng)建訂閱組，建議線下開啟，線上關(guān)閉autoCreateSubscriptionGroup=true# broker對外服務(wù)的監(jiān)聽端口listenPort=11011# abort文件存儲(chǔ)路徑abortFile=/data/rocketmq/store-slave/abort# 消息存儲(chǔ)路徑storePathRootDir=/data/rocketmq/store-slave# commitLog存儲(chǔ)路徑storePathCommitLog=/data/rocketmq/store-slave/commitlog# 消費(fèi)隊(duì)列存儲(chǔ)路徑storePathConsumeQueue=/data/rocketmq/store-slave/consumequeue# 消息索引存儲(chǔ)路徑storePathIndex=/data/rocketmq/store-slave/index# checkpoint文件存儲(chǔ)路徑storeCheckpoint=/data/rocketmq/store-slave/checkpoint# 限制的消息大小maxMessageSize=655361234567891011121314151617181920212223242526272829303132333435363738394041424344

3.3.5、啟動(dòng)broker

在服務(wù)器1中啟動(dòng)broker-a（master）和broker-b-s（slave）

cd /opt/home/rocketmq4.8.0/binnohup ./mqbroker -c ../conf/2m-2s-async/broker-a.properties > logs/broker-a.out 2>1 &nohup ./mqbroker -c ../conf/2m-2s-async/broker-b-s.properties > logs/broker-b-s.out 2>1 &123

在服務(wù)器2中啟動(dòng)broker-b（master）和broker-a-s（slave）

cd /opt/home/rocketmq4.8.0/binnohup ./mqbroker -c ../conf/2m-2s-async/broker-b.properties > logs/broker-b.out 2>1 &nohup ./mqbroker -c ../conf/2m-2s-async/broker-a-s.properties > logs/broker-a-s.out 2>1 &123

3.4、驗(yàn)證集群

使用RocketMQ提供的tools工具驗(yàn)證集群是否正常工作。

在服務(wù)器1上配置環(huán)境變量，用于被tools中的生產(chǎn)者和消費(fèi)者程序讀取該變量。

## 配置環(huán)境變量export NAMESRV_ADDR='192.168.0.101:9876;192.168.0.102:9876;192.168.0.103:9876'## 環(huán)境變量生效source /etc/profile12345

啟動(dòng)生產(chǎn)者

./tools.sh org.apache.rocketmq.example.quickstart.Producer1

啟動(dòng)消費(fèi)者

./tools.sh org.apache.rocketmq.example.quickstart.Consumer1

四、mqadmin管理工具

RocketMQ提供了命令工具用于管理topic、broker、集群、消息等。比如可以使用mqadmin創(chuàng)建topic：

該命令在官網(wǎng)中有詳細(xì)的用法解釋：https://github.com/apache/rocketmq/tree/master/docs/cn

./mqadmin updateTopic -n 192.168.0.101:9876 -c DefaultCluster -t myTopic11

下面提供了mqadmin工具的各種命令。

4.1、創(chuàng)建topic：updateTopic

4.2、刪除Topic：deleteTopic

4.3、創(chuàng)建（修訂）訂閱組：updateSubGroup

4.4、刪除訂閱組配置：deleteSubGroup

4.5、更新Broker 配置?件：updateBrokerConfig

4.6、查看Topic 列表信息：topicList

4.7、查看Topic 路由信息：topicRoute

4.8、查看Topic 統(tǒng)計(jì)信息：topicStats

4.9、查看Broker 統(tǒng)計(jì)信息：brokerStats

4.10、根據(jù)消息ID 查詢消息：queryMsgById

4.11、根據(jù)消息Key 查詢消息：queryMsgByKey

4.12、根據(jù)Offset 查詢消息：queryMsgByOffset

4.13、查詢Producer 的網(wǎng)絡(luò)連接：producerConnection

4.14、查詢Consumer 的?絡(luò)連接：consumerConnection

4.15、查看訂閱組消費(fèi)狀態(tài)：consumerProgress

4.16、查看集群消息：clusterList

4.18、添加（更新）KV 配置信息：updateKvConfig

4.19、刪除KV 配置信息：deleteKvConfig

4.20、添加（更新）Project group 配置信息：updateProjectGroup

4.21、刪除Project group 配置信息：deleteProjectGroup

4.22、取得Project group 配置信息：getProjectGroup

4.23、設(shè)置消費(fèi)進(jìn)度：resetOffsetByTime

根據(jù)時(shí)間來設(shè)置消費(fèi)進(jìn)度，設(shè)置之前要關(guān)閉這個(gè)訂閱組的所有consumer，設(shè)置完再啟動(dòng)，方可生效。

4.24、清除特定Broker權(quán)限：wipeWritePerm

4.25、獲取Consumer消費(fèi)進(jìn)度：getConsumerStatus

該命令只打印當(dāng)前與cluster 連接的consumer 的消費(fèi)進(jìn)度

五、安裝可視化管理控制平臺(tái)

按照1.4進(jìn)行操作，修改其src/main/resources中的application.properties配置文件。

server.contextPath=
server.port=8080
#spring.application.index=true
spring.application.name=rocketmq-console
spring.http.encoding.charset=UTF-8
spring.http.encoding.enabled=true
spring.http.encoding.force=true
logging.config=classpath:logback.xml
#if this value is empty,use env value rocketmq.config.namesrvAddr  NAMESRV_ADDR | now, you can set it in ops page.default localhost:9876
rocketmq.config.namesrvAddr=192.168.0.101:9876;192.168.0.102:9876;192.168.0.103:9876
#if you use rocketmq version < 3.5.8, rocketmq.config.isVIPChannel should be false.default true
rocketmq.config.isVIPChannel=
#rocketmq-console's data path:dashboard/monitor
rocketmq.config.dataPath=/tmp/rocketmq-console/data
#set it false if you don't want use dashboard.default true
rocketmq.config.enableDashBoardCollect=true12345678910111213141516

在rocketmq-console目錄下，打開cmd窗口運(yùn)行maven的打包命令。

mvn clean package -Dmaven.test.skip=true1

這里為了方便，我將jar包上傳到服務(wù)器，然后啟動(dòng)。

nohup java -jar rocketmq-console-ng-1.0.0.jar > rocketmq-console.out 2>1 &1

訪問地址：http://192.168.0.101:8080/#/