RocketMQ安裝部署

• 一、單機部署

• 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、發送/接收消息測試

• 1.3.4、關閉服務

• 1.2.1、runserver.sh

• 1.2.2、runbroker.sh

• 1.2.3、tools.sh

• 1.2.4、自動動創建Topic功能

• 1.1、下載

• 1.2、修改RocketMQ啟動配置

• 1.3、RocketMQ 啟動

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

• 二、集群搭建理論

• 2.1.1、復制策略

• 2.1.2、刷盤策略

• 2.1、數據復制與刷盤策略

• 2.2、Broker集群模式

• 三、搭建主從異步集群

• 3.3.1、broker-a的master節點

• 3.3.2、broker-a的slave節點

• 3.3.3、broker-b的master節點

• 3.3.4、broker-b的slave節點

• 3.3.5、啟動broker

• 3.1、集群架構

• 3.2、啟動三臺nameserver

• 3.3、配置broker

• 3.4、驗證集群

• 四、mqadmin管理工具

• 4.1、創建topic：updateTopic

• 4.2、刪除Topic：deleteTopic

• 4.3、創建（修訂）訂閱組：updateSubGroup

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

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

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

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

• 4.8、查看Topic 統計信息：topicStats

• 4.9、查看Broker 統計信息：brokerStats

• 4.10、根據消息ID 查詢消息：queryMsgById

• 4.11、根據消息Key 查詢消息：queryMsgByKey

• 4.12、根據Offset 查詢消息：queryMsgByOffset

• 4.13、查詢Producer 的網絡連接：producerConnection

• 4.14、查詢Consumer 的?絡連接：consumerConnection

• 4.15、查看訂閱組消費狀態：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、設置消費進度：resetOffsetByTime

• 4.24、清除特定Broker權限：wipeWritePerm

• 4.25、獲取Consumer消費進度：getConsumerStatus

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

一、單機部署

環境說明

對于 RocketMQ 4.3.0版本，官方要求環境如下，其中 Git 用于從 GitHub 獲取源碼，沒有安裝也沒關系，可以直接下載
官方推薦的流程是：Linux 系統上安裝 Git 工具 、Maven、Java JDK
Git 工具用于直接從 GitHub 獲取 RocketMQ 項目源碼下載到 Linux 系統上
然后 Maven 將 RocketMQ 源碼進行編譯成二進制文件
安裝了 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、自動動創建Topic功能

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

autoCreateTopicEnable=true1

1.3、RocketMQ 啟動

1.3.1、啟動 NameServer

進入 RocketMQ 安裝目錄下的/opt/home/rocketmq4.8.0/bin目錄進行操作

cd /opt/home/rocketmq4.8.0/bin1

執行命令啟動NameServer

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

查看啟動狀態，在當前目錄下會有一個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環境變量# 需要 export 環境變量12345

1.3.2、啟動 Broker

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

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

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

查看啟動狀態，啟動之后同樣提示將日志信息追加到了當前目錄下的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、發送/接收消息測試

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

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

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

• 環境變量，如：NAMESRV_ADDR

• HTTP 端點

發送消息

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

消息發送完畢之后就會退出，在同一窗口中可以使用消費者類來進行接收消息，消費是多線程的。

1.3.4、關閉服務

與啟動順序相反進行關閉，先關閉 broker、在關閉 nameserv

./mqshutdown broker
./mqshutdown namesrv12

1.4、控制臺的安裝與啟動

RocketMQ有一個可視化的dashboard，通過該控制臺可以直觀的查看到很多數據。

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地址，集群環境為集群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技術，是一個業界標準，是一 項可以根據XML Schema生成Java類的技術。

    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包上傳到服務器，然后啟動。

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、數據復制與刷盤策略

2.1.1、復制策略

復制策略是Broker的Master與Slave間的數據同步方式。分為同步復制與異步復制：

• 同步復制：消息寫入master后，master會等待slave同步數據成功后才向producer返回成功ACK。

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

異步復制策略會降低系統的寫入延遲，RT變小，提高了系統的吞吐量。

2.1.2、刷盤策略

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

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

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

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

2.2、Broker集群模式

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

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

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

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

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

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

三、搭建主從異步集群

3.1、集群架構

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

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

3.2、啟動三臺nameserver

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

在每臺服務器的bin目錄下執行如下命令：

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

服務器1

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

服務器2

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

服務器3

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

3.3、配置broker

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

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

3.3.1、broker-a的master節點

在服務器1上，進?到conf/2m-2s-async?件夾內，修改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所在服務器的ipbrokerIP1=192.168.0.101# broker的id，0表示master，>0表示slavebrokerId=0# 指定刪除消息存儲過期文件的時間為凌晨4點deleteWhen=04# 指定未發生更新的消息存儲文件的保留時長為48小時，48小時后過期，將會被刪除fileReservedTime=48# 指定當前broker為異步復制masterbrokerRole=ASYNC_MASTER# 指定刷盤策略為異步刷盤flushDiskType=ASYNC_FLUSH# 指定Name Server的地址namesrvAddr=192.168.0.101:9876;192.168.0.102:9876;192.168.0.103:9876# 在發送消息?動創建不存在的topic時，默認創建的隊列數為4個defaultTopicQueueNums=4# 是否允許 Broker 自動創建Topic，建議線下開啟，線上關閉autoCreateTopicEnable=true# 是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉autoCreateSubscriptionGroup=true# broker對外服務的監聽端口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節點

在服務器2上，進?到conf/2m-2s-async文件夾內，修改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所在服務器的ipbrokerIP1=192.168.0.102# broker的id，0表示master，>0表示slavebrokerId=1# 指定刪除消息存儲過期文件的時間為凌晨4點deleteWhen=04# 指定未發生更新的消息存儲文件的保留時長為48小時，48小時后過期，將會被刪除fileReservedTime=48# 指定當前broker為異步復制masterbrokerRole=SLAVE# 指定刷盤策略為異步刷盤flushDiskType=ASYNC_FLUSH# 指定Name Server的地址namesrvAddr=192.168.0.101:9876;192.168.0.102:9876;192.168.0.103:9876# 在發送消息?動創建不存在的topic時，默認創建的隊列數為4個defaultTopicQueueNums=4# 是否允許 Broker 自動創建Topic，建議線下開啟，線上關閉autoCreateTopicEnable=true# 是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉autoCreateSubscriptionGroup=true# broker對外服務的監聽端口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節點

在服務器2上，進入到conf/2m-2s-async文件夾內，修改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所在服務器的ipbrokerIP1=192.168.0.102# broker的id，0表示master，>0表示slavebrokerId=0# 指定刪除消息存儲過期文件的時間為凌晨4點deleteWhen=04# 指定未發生更新的消息存儲文件的保留時長為48小時，48小時后過期，將會被刪除fileReservedTime=48# 指定當前broker為異步復制masterbrokerRole=ASYNC_MASTER# 指定刷盤策略為異步刷盤flushDiskType=ASYNC_FLUSH# 指定Name Server的地址namesrvAddr=192.168.0.101:9876;192.168.0.102:9876;192.168.0.103:9876# 在發送消息?動創建不存在的topic時，默認創建的隊列數為4個defaultTopicQueueNums=4# 是否允許 Broker 自動創建Topic，建議線下開啟，線上關閉autoCreateTopicEnable=true# 是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉autoCreateSubscriptionGroup=true# broker對外服務的監聽端口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節點

在服務器1上，進入到conf/2m-2s-async文件夾內，修改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所在服務器的ipbrokerIP1=192.168.0.101# broker的id，0表示master，>0表示slavebrokerId=1# 指定刪除消息存儲過期文件的時間為凌晨4點deleteWhen=04# 指定未發生更新的消息存儲文件的保留時長為48小時，48小時后過期，將會被刪除fileReservedTime=48# 指定當前broker為異步復制masterbrokerRole=SLAVE# 指定刷盤策略為異步刷盤flushDiskType=ASYNC_FLUSH# 指定Name Server的地址namesrvAddr=192.168.0.101:9876;192.168.0.102:9876;192.168.0.103:9876# 在發送消息?動創建不存在的topic時，默認創建的隊列數為4個defaultTopicQueueNums=4# 是否允許 Broker 自動創建Topic，建議線下開啟，線上關閉autoCreateTopicEnable=true# 是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉autoCreateSubscriptionGroup=true# broker對外服務的監聽端口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

在服務器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

在服務器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工具驗證集群是否正常工作。

在服務器1上配置環境變量，用于被tools中的生產者和消費者程序讀取該變量。

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

啟動生產者

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

啟動消費者

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

四、mqadmin管理工具

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

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

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

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

4.1、創建topic：updateTopic

4.2、刪除Topic：deleteTopic

4.3、創建（修訂）訂閱組：updateSubGroup

4.4、刪除訂閱組配置：deleteSubGroup

4.5、更新Broker 配置?件：updateBrokerConfig

4.6、查看Topic 列表信息：topicList

4.7、查看Topic 路由信息：topicRoute

4.8、查看Topic 統計信息：topicStats

4.9、查看Broker 統計信息：brokerStats

4.10、根據消息ID 查詢消息：queryMsgById

4.11、根據消息Key 查詢消息：queryMsgByKey

4.12、根據Offset 查詢消息：queryMsgByOffset

4.13、查詢Producer 的網絡連接：producerConnection

4.14、查詢Consumer 的?絡連接：consumerConnection

4.15、查看訂閱組消費狀態：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、設置消費進度：resetOffsetByTime

根據時間來設置消費進度，設置之前要關閉這個訂閱組的所有consumer，設置完再啟動，方可生效。

4.24、清除特定Broker權限：wipeWritePerm

4.25、獲取Consumer消費進度：getConsumerStatus

該命令只打印當前與cluster 連接的consumer 的消費進度

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

按照1.4進行操作，修改其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包上傳到服務器，然后啟動。

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

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