對于 RocketMQ 4.3.0版本,官方要求環(huán)境如下,其中 Git 用于從 GitHub 獲取源碼,沒有安裝也沒關(guān)系,可以直接下載
環(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 了
## 切換/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
修改目錄 /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
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
vi runbroker.sh# JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"1234
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
在conf/broker.conf
?件中加?如下配置,開啟自動(dòng)動(dòng)創(chuàng)建Topic功能,不建議開啟。
autoCreateTopicEnable=true1
進(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
同樣進(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
發(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)是多線程的。
與啟動(dòng)順序相反進(jìn)行關(guān)閉,先關(guān)閉 broker、在關(guān)閉 nameserv
./mqshutdown broker ./mqshutdown namesrv12
RocketMQ有一個(gè)可視化的dashboard,通過該控制臺(tái)可以直觀的查看到很多數(shù)據(jù)。
下載地址: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
使用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
在解壓目錄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
在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/#/
更換語言為中文
復(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)的吞吐量。
刷盤策略指的是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)行落盤。
為了追求更好的性能,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?可用集群。
這里要搭建一個(gè)雙主雙從異步復(fù)制的Broker集群。為了方便,這里使用了兩臺(tái)主機(jī)來完成集群的搭建。 這兩臺(tái)主機(jī)的功能與broker角色分配如下表。
服務(wù)器 | IP | NameServer | BROKER角色 |
---|---|---|---|
服務(wù)器1 | 192.168.0.101 | 192.168.0.101:9876 | broker-a(master),broker-b-s(slave) |
服務(wù)器2 | 192.168.0.102 | 192.168.0.102:9876 | broker-b(master),broker-a-s(slave) |
服務(wù)器3 | 192.168.0.103 | 192.168.0.103:9876 |
三臺(tái)服務(wù)器都需要安裝JDK和RocketMQ,安裝步驟參考如上。
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
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保存路徑不能?樣。
在服務(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
在服務(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
在服務(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
在服務(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
在服務(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
使用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
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
參數(shù) | 是否必填 | 說明 |
---|---|---|
-b | 如果-c為空,則必填 | broker 地址,表示topic 建在該broker |
-c | 如果-b為空,則必填 | cluster 名稱,表示topic 建在該集群(集群可通過clusterList 查詢) |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
-p | 否 | 指定新topic 的權(quán)限限制( W|R|WR ) |
-r | 否 | 可讀隊(duì)列數(shù)(默認(rèn)為8) |
-w | 否 | 可寫隊(duì)列數(shù)(默認(rèn)為8) |
-t | 是 | topic 名稱(名稱只能使用字符 1+$ ) |
參數(shù) | 是否必填 | 說明 |
---|---|---|
-c | 是 | cluster 名稱,表示topic 建在該集群(集群可通過clusterList 查詢) |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
-t | 是 | topic 名稱(名稱只能使用字符 2+$ ) |
4.3、創(chuàng)建(修訂)訂閱組:updateSubGroup
參數(shù) | 是否必填 | 說明 |
---|---|---|
-b | 如果-c為空,則必填 | broker 地址,表示topic 建在該broker |
-c | 如果-b為空,則必填 | cluster 名稱,表示topic 建在該集群(集群可通過clusterList 查詢) |
-d | 否 | 是否容許廣播方式消費(fèi) |
-g | 是 | 訂閱組名 |
-i | 否 | 從哪個(gè)broker 開始消費(fèi) |
-m | 否 | 是否容許從隊(duì)列的最小位置開始消費(fèi),默認(rèn)會(huì)設(shè)置為false |
-q | 否 | 消費(fèi)失敗的消息放到?個(gè)重試隊(duì)列,每個(gè)訂閱組配置幾個(gè)重試隊(duì)列 |
-r | 否 | 重試消費(fèi)最大次數(shù),超過則投遞到死信隊(duì)列,不再投遞,并報(bào)警 |
-s | 否 | 消費(fèi)功能是否開啟 |
-w | 否 | 發(fā)現(xiàn)消息堆積后后,將Consumer 的消費(fèi)請求重定向到另外?臺(tái)Slave 機(jī)器 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
參數(shù) | 是否必填 | 說明 |
---|---|---|
-b | 如果-c為空,則必填 | broker 地址,表示topic 建在該broker |
-c | 如果-b為空,則必填 | cluster 名稱,表示topic 建在該集群(集群可通過clusterList 查詢) |
-g | 是 | 訂閱組名 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.5、更新Broker 配置?件:updateBrokerConfig
參數(shù) | 是否必填 | 說明 |
---|---|---|
-b | 如果-c為空,則必填 | broker 地址,表示topic 建在該broker |
-c | 如果-b為空,則必填 | cluster 名稱,表示topic 建在該集群(集群可通過clusterList 查詢) |
-k | 是 | key 值 |
-v | 否 | value 值 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
參數(shù) | 是否必填 | 說明 |
---|---|---|
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
參數(shù) | 是否必填 | 說明 |
---|---|---|
-t | 是 | topic 名稱 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.8、查看Topic 統(tǒng)計(jì)信息:topicStats
參數(shù) | 是否必填 | 說明 |
---|---|---|
-t | 是 | topic 名稱 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.9、查看Broker 統(tǒng)計(jì)信息:brokerStats
參數(shù) | 是否必填 | 說明 |
---|---|---|
-b | 是 | broker 地址 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.10、根據(jù)消息ID 查詢消息:queryMsgById
參數(shù) | 是否必填 | 說明 |
---|---|---|
-i | 是 | 消息id |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.11、根據(jù)消息Key 查詢消息:queryMsgByKey
參數(shù) | 是否必填 | 說明 |
---|---|---|
-f | 否 | 被查詢消息的止時(shí)間 |
-k | 是 | msgKey |
-t | 是 | topic 名稱 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.12、根據(jù)Offset 查詢消息:queryMsgByOffset
參數(shù) | 是否必填 | 說明 |
---|---|---|
-b | 是 | Broker 名稱,表示訂閱組建在該broker(這里需要注意填寫的是 broker 的名稱,不是broker 的地址,broker名稱可以在clusterList查到) |
-i | 是 | query 隊(duì)列id |
-o | 是 | offset 值 |
-t | 是 | topic 名稱 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.13、查詢Producer 的網(wǎng)絡(luò)連接:producerConnection
參數(shù) | 是否必填 | 說明 |
---|---|---|
-g | 是 | 生產(chǎn)者所屬組名 |
-t | 是 | topic 名稱 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.14、查詢Consumer 的?絡(luò)連接:consumerConnection
參數(shù) | 是否必填 | 說明 |
---|---|---|
-g | 是 | 生產(chǎn)者所屬組名 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.15、查看訂閱組消費(fèi)狀態(tài):consumerProgress
參數(shù) | 是否必填 | 說明 |
---|---|---|
-g | 是 | 生產(chǎn)者所屬組名 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
參數(shù) | 是否必填 | 說明 |
---|---|---|
-m | 否 | 打印更多信息 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.18、添加(更新)KV 配置信息:updateKvConfig
參數(shù) | 是否必填 | 說明 |
---|---|---|
-k | 是 | key 值 |
-v | 是 | value 值 |
-s | 是 | Namespace 值 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
參數(shù) | 是否必填 | 說明 |
---|---|---|
-k | 是 | key 值 |
-s | 是 | Namespace 值 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.20、添加(更新)Project group 配置信息:updateProjectGroup
參數(shù) | 是否必填 | 說明 |
---|---|---|
-p | 是 | project group 名 |
-i | 否 | 服務(wù)器ip |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.21、刪除Project group 配置信息:deleteProjectGroup
參數(shù) | 是否必填 | 說明 |
---|---|---|
-p | 是 | project group 名 |
-i | 否 | 服務(wù)器ip |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.22、取得Project group 配置信息:getProjectGroup
參數(shù) | 是否必填 | 說明 |
---|---|---|
-p | 是 | project group 名 |
-i | 否 | 服務(wù)器ip |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.23、設(shè)置消費(fèi)進(jìn)度:resetOffsetByTime
根據(jù)時(shí)間來設(shè)置消費(fèi)進(jìn)度,設(shè)置之前要關(guān)閉這個(gè)訂閱組的所有consumer,設(shè)置完再啟動(dòng),方可生效。
參數(shù) | 是否必填 | 說明 |
---|---|---|
-f | 否 | 通過時(shí)間戳強(qiáng)制回滾(true|false),默認(rèn)為true |
-s | 是 | 時(shí)間戳 |
-g | 是 | 消費(fèi)者所屬組名 |
-t | 是 | topic 名稱 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.24、清除特定Broker權(quán)限:wipeWritePerm
參數(shù) | 是否必填 | 說明 |
---|---|---|
-b | 是 | broker 地址 |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
4.25、獲取Consumer消費(fèi)進(jìn)度:getConsumerStatus
該命令只打印當(dāng)前與cluster 連接的consumer 的消費(fèi)進(jìn)度
參數(shù) | 是否必填 | 說明 |
---|---|---|
-g | 是 | 消費(fèi)者所屬組名 |
-t | 是 | 查詢主題 |
-i | 否 | Consumer 客戶端ip |
-h | 否 | 打印幫助 |
-n | 是 | nameserve 服務(wù)地址列表,格式ip:port;ip:port;… |
按照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/#/
來源:本文內(nèi)容搜集或轉(zhuǎn)自各大網(wǎng)絡(luò)平臺(tái),并已注明來源、出處,如果轉(zhuǎn)載侵犯您的版權(quán)或非授權(quán)發(fā)布,請聯(lián)系小編,我們會(huì)及時(shí)審核處理。
聲明:江蘇教育黃頁對文中觀點(diǎn)保持中立,對所包含內(nèi)容的準(zhǔn)確性、可靠性或者完整性不提供任何明示或暗示的保證,不對文章觀點(diǎn)負(fù)責(zé),僅作分享之用,文章版權(quán)及插圖屬于原作者。
Copyright?2013-2024 JSedu114 All Rights Reserved. 江蘇教育信息綜合發(fā)布查詢平臺(tái)保留所有權(quán)利
蘇公網(wǎng)安備32010402000125
蘇ICP備14051488號-3技術(shù)支持:南京博盛藍(lán)睿網(wǎng)絡(luò)科技有限公司
南京思必達(dá)教育科技有限公司版權(quán)所有 百度統(tǒng)計(jì)