RocketMQ安裝部署

• 一、單機(jī)部署

• 1.4.1、下載

• 1.4.2、修改配置

• 1.4.3、添加依賴

• 1.4.4、打包

• 1.4.5、啟動

• 1.4.6、訪問

• 1.3.1、啟動 NameServer

• 1.3.2、啟動 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、自動動創(chuàng)建Topic功能

• 1.1、下載

• 1.2、修改RocketMQ啟動配置

• 1.3、RocketMQ 啟動

• 1.4、控制臺的安裝與啟動

• 二、集群搭建理論

• 2.1.1、復(fù)制策略

• 2.1.2、刷盤策略

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

• 2.2、Broker集群模式

• 三、搭建主從異步集群

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

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

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

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

• 3.3.5、啟動broker

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

• 3.2、啟動三臺nameserver

• 3.3、配置broker

• 3.4、驗證集群

• 四、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)計信息：topicStats

• 4.9、查看Broker 統(tǒng)計信息：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、查看訂閱組消費狀態(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è)置消費進(jìn)度：resetOffsetByTime

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

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

• 五、安裝可視化管理控制平臺

一、單機(jī)部署

環(huán)境說明

對于 RocketMQ 4.3.0版本，官方要求環(huán)境如下，其中 Git 用于從 GitHub 獲取源碼，沒有安裝也沒關(guān)系，可以直接下載
官方推薦的流程是：Linux 系統(tǒng)上安裝 Git 工具 、Maven、Java JDK
Git 工具用于直接從 GitHub 獲取 RocketMQ 項目源碼下載到 Linux 系統(tǒng)上
然后 Maven 將 RocketMQ 源碼進(jìn)行編譯成二進(jìn)制文件
安裝了 Java JDK 就可以運行 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## 項目解壓unzip rocketmq-all-4.8.0-bin-release.zip## 重命名mv rocketmq-all-4.8.0-bin-release rocketmq4.8.01234567891011

1.2、修改RocketMQ啟動配置

修改目錄 /opt/home/rocketmq4.8.0/bin 下的 3 個配置文件： runserver.sh、runbroker.sh 、tools.sh不然會報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、自動動創(chuàng)建Topic功能

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

autoCreateTopicEnable=true1

1.3、RocketMQ 啟動

1.3.1、啟動 NameServer

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

cd /opt/home/rocketmq4.8.0/bin1

執(zhí)行命令啟動NameServer

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

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

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

解決報錯

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

1.3.2、啟動 Broker

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

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

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

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

## 查看日志tail -f logs/mqbroker.out## 看到以下表示啟動成功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 提供了多種方法來實現(xiàn)這一點

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

• Java 選項，如：rocketmq.namesrv.addr

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

• HTTP 端點

發(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ā)送完畢之后就會退出，在同一窗口中可以使用消費者類來進(jìn)行接收消息，消費是多線程的。

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

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

./mqshutdown broker
./mqshutdown namesrv12

1.4、控制臺的安裝與啟動

RocketMQ有一個可視化的dashboard，通過該控制臺可以直觀的查看到很多數(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打開項目，修改其src/main/resources中的application.properties配置文件。

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

• 指定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ù)，是一個業(yè)界標(biāo)準(zhǔn)，是一 項可以根據(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窗口運行maven的打包命令。

mvn clean package -Dmaven.test.skip=true1

1.4.5、啟動

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

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會等待slave同步數(shù)據(jù)成功后才向producer返回成功ACK。

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

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

2.1.2、刷盤策略

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

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

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

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

2.2、Broker集群模式

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

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

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

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

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

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

三、搭建主從異步集群

3.1、集群架構(gòu)

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

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

3.2、啟動三臺nameserver

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

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

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

服務(wù)器1

## 后臺靜默啟動nohup ./mqnamesrv -n 192.168.0.101:9876 > logs/mqnamesrv.out 2>1 &12

服務(wù)器2

## 后臺靜默啟動nohup ./mqnamesrv -n 192.168.0.102:9876 > logs/mqnamesrv.out 2>1 &12

服務(wù)器3

## 后臺靜默啟動nohup ./mqnamesrv -n 192.168.0.103:9876 > logs/mqnamesrv.out 2>1 &12

3.3、配置broker

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

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

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

在服務(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# 指定整個broker集群的名稱，或者說是RocketMQ集群的名稱brokerClusterName=DefaultCluster# 指定master-slave集群的名稱。一個RocketMQ集群可以包含多個master-slave集群brokerName=broker-a# broker所在服務(wù)器的ipbrokerIP1=192.168.0.101# broker的id，0表示master，>0表示slavebrokerId=0# 指定刪除消息存儲過期文件的時間為凌晨4點deleteWhen=04# 指定未發(fā)生更新的消息存儲文件的保留時長為48小時，48小時后過期，將會被刪除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ā)送消息?動創(chuàng)建不存在的topic時，默認(rèn)創(chuàng)建的隊列數(shù)為4個defaultTopicQueueNums=4# 是否允許 Broker 自動創(chuàng)建Topic，建議線下開啟，線上關(guān)閉autoCreateTopicEnable=true# 是否允許 Broker 自動創(chuàng)建訂閱組，建議線下開啟，線上關(guān)閉autoCreateSubscriptionGroup=true# broker對外服務(wù)的監(jiān)聽端口listenPort=10911# abort文件存儲路徑abortFile=/data/rocketmq/store/abort# 消息存儲路徑storePathRootDir=/data/rocketmq/store# commitLog存儲路徑storePathCommitLog=/data/rocketmq/store/commitlog# 消費隊列存儲路徑storePathConsumeQueue=/data/rocketmq/store/consumequeue# 消息索引存儲路徑storePathIndex=/data/rocketmq/store/index# checkpoint文件存儲路徑storeCheckpoint=/data/rocketmq/store/checkpoint# 限制的消息大小maxMessageSize=655361234567891011121314151617181920212223242526272829303132333435363738394041424344

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

在服務(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# 指定整個broker集群的名稱，或者說是RocketMQ集群的名稱brokerClusterName=DefaultCluster# 指定master-slave集群的名稱。一個RocketMQ集群可以包含多個master-slave集群brokerName=broker-a# broker所在服務(wù)器的ipbrokerIP1=192.168.0.102# broker的id，0表示master，>0表示slavebrokerId=1# 指定刪除消息存儲過期文件的時間為凌晨4點deleteWhen=04# 指定未發(fā)生更新的消息存儲文件的保留時長為48小時，48小時后過期，將會被刪除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ā)送消息?動創(chuàng)建不存在的topic時，默認(rèn)創(chuàng)建的隊列數(shù)為4個defaultTopicQueueNums=4# 是否允許 Broker 自動創(chuàng)建Topic，建議線下開啟，線上關(guān)閉autoCreateTopicEnable=true# 是否允許 Broker 自動創(chuàng)建訂閱組，建議線下開啟，線上關(guān)閉autoCreateSubscriptionGroup=true# broker對外服務(wù)的監(jiān)聽端口listenPort==11011# abort文件存儲路徑abortFile=/data/rocketmq/store-slave/abort# 消息存儲路徑storePathRootDir=/data/rocketmq/store-slave# commitLog存儲路徑storePathCommitLog=/data/rocketmq/store-slave/commitlog# 消費隊列存儲路徑storePathConsumeQueue=/data/rocketmq/store-slave/consumequeue# 消息索引存儲路徑storePathIndex=/data/rocketmq/store-slave/index# checkpoint文件存儲路徑storeCheckpoint=/data/rocketmq/store-slave/checkpoint# 限制的消息大小maxMessageSize=655361234567891011121314151617181920212223242526272829303132333435363738394041424344

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

在服務(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# 指定整個broker集群的名稱，或者說是RocketMQ集群的名稱brokerClusterName=DefaultCluster# 指定master-slave集群的名稱。一個RocketMQ集群可以包含多個master-slave集群brokerName=broker-b# broker所在服務(wù)器的ipbrokerIP1=192.168.0.102# broker的id，0表示master，>0表示slavebrokerId=0# 指定刪除消息存儲過期文件的時間為凌晨4點deleteWhen=04# 指定未發(fā)生更新的消息存儲文件的保留時長為48小時，48小時后過期，將會被刪除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ā)送消息?動創(chuàng)建不存在的topic時，默認(rèn)創(chuàng)建的隊列數(shù)為4個defaultTopicQueueNums=4# 是否允許 Broker 自動創(chuàng)建Topic，建議線下開啟，線上關(guān)閉autoCreateTopicEnable=true# 是否允許 Broker 自動創(chuàng)建訂閱組，建議線下開啟，線上關(guān)閉autoCreateSubscriptionGroup=true# broker對外服務(wù)的監(jiān)聽端口listenPort=10911# abort文件存儲路徑abortFile=/data/rocketmq/store/abort# 消息存儲路徑storePathRootDir=/data/rocketmq/store# commitLog存儲路徑storePathCommitLog=/data/rocketmq/store/commitlog# 消費隊列存儲路徑storePathConsumeQueue=/data/rocketmq/store/consumequeue# 消息索引存儲路徑storePathIndex=/data/rocketmq/store/index# checkpoint文件存儲路徑storeCheckpoint=/data/rocketmq/store/checkpoint# 限制的消息大小maxMessageSize=655361234567891011121314151617181920212223242526272829303132333435363738394041424344

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

在服務(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# 指定整個broker集群的名稱，或者說是RocketMQ集群的名稱brokerClusterName=DefaultCluster# 指定master-slave集群的名稱。一個RocketMQ集群可以包含多個master-slave集群brokerName=broker-b# broker所在服務(wù)器的ipbrokerIP1=192.168.0.101# broker的id，0表示master，>0表示slavebrokerId=1# 指定刪除消息存儲過期文件的時間為凌晨4點deleteWhen=04# 指定未發(fā)生更新的消息存儲文件的保留時長為48小時，48小時后過期，將會被刪除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ā)送消息?動創(chuàng)建不存在的topic時，默認(rèn)創(chuàng)建的隊列數(shù)為4個defaultTopicQueueNums=4# 是否允許 Broker 自動創(chuàng)建Topic，建議線下開啟，線上關(guān)閉autoCreateTopicEnable=true# 是否允許 Broker 自動創(chuàng)建訂閱組，建議線下開啟，線上關(guān)閉autoCreateSubscriptionGroup=true# broker對外服務(wù)的監(jiān)聽端口listenPort=11011# abort文件存儲路徑abortFile=/data/rocketmq/store-slave/abort# 消息存儲路徑storePathRootDir=/data/rocketmq/store-slave# commitLog存儲路徑storePathCommitLog=/data/rocketmq/store-slave/commitlog# 消費隊列存儲路徑storePathConsumeQueue=/data/rocketmq/store-slave/consumequeue# 消息索引存儲路徑storePathIndex=/data/rocketmq/store-slave/index# checkpoint文件存儲路徑storeCheckpoint=/data/rocketmq/store-slave/checkpoint# 限制的消息大小maxMessageSize=655361234567891011121314151617181920212223242526272829303132333435363738394041424344

3.3.5、啟動broker

在服務(wù)器1中啟動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中啟動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、驗證集群

使用RocketMQ提供的tools工具驗證集群是否正常工作。

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

## 配置環(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

啟動生產(chǎn)者

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

啟動消費者

./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)計信息：topicStats

4.9、查看Broker 統(tǒng)計信息：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、查看訂閱組消費狀態(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è)置消費進(jìn)度：resetOffsetByTime

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

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

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

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

五、安裝可視化管理控制平臺

按照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窗口運行maven的打包命令。

mvn clean package -Dmaven.test.skip=true1

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

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

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