?
1. MQ介紹
##1.1 為什么要用MQ
消息隊列是一種“先進先出”的數據結構
轉存失敗重新上傳取消
其應用場景主要包含以下3個方面
- 應用解耦
系統的耦合性越高，容錯性就越低。以電商應用為例，用戶創建訂單后，如果耦合調用庫存系統、物流系統、支付系統，任何一個子系統出了故障或者因為升級等原因暫時不可用，都會造成下單操作異常，影響用戶使用體驗。
轉存失敗重新上傳取消轉存失敗重新上傳取消
使用消息隊列解耦合，系統的耦合性就會提高了。比如物流系統發生故障，需要幾分鐘才能來修復，在這段時間內，物流系統要處理的數據被緩存到消息隊列中，用戶的下單操作正常完成。當物流系統回復后，補充處理存在消息隊列中的訂單消息即可，終端系統感知不到物流系統發生過幾分鐘故障。
轉存失敗重新上傳取消轉存失敗重新上傳取消
- 流量削峰
轉存失敗重新上傳取消
應用系統如果遇到系統請求流量的瞬間猛增，有可能會將系統壓垮。有了消息隊列可以將大量請求緩存起來，分散到很長一段時間處理，這樣可以大大提到系統的穩定性和用戶體驗。
轉存失敗重新上傳取消
一般情況，為了保證系統的穩定性，如果系統負載超過閾值，就會阻止用戶請求，這會影響用戶體驗，而如果使用消息隊列將請求緩存起來，等待系統處理完畢后通知用戶下單完畢，這樣總不能下單體驗要好。
處于經濟考量目的：
業務系統正常時段的QPS如果是1000，流量最高峰是10000，為了應對流量高峰配置高性能的服務器顯然不劃算，這時可以使用消息隊列對峰值流量削峰
- 數據分發
轉存失敗重新上傳取消轉存失敗重新上傳取消
通過消息隊列可以讓數據在多個系統更加之間進行流通。數據的產生方不需要關心誰來使用數據，只需要將數據發送到消息隊列，數據使用方直接在消息隊列中直接獲取數據即可
1.2 MQ的優點和缺點
優點：解耦、削峰、數據分發
缺點包含以下幾點：
-
系統可用性降低
系統引入的外部依賴越多，系統穩定性越差。一旦MQ宕機，就會對業務造成影響。
如何保證MQ的高可用？
-
系統復雜度提高
MQ的加入大大增加了系統的復雜度，以前系統間是同步的遠程調用，現在是通過MQ進行異步調用。
如何保證消息沒有被重復消費？怎么處理消息丟失情況？那么保證消息傳遞的順序性？
-
一致性問題
A系統處理完業務，通過MQ給B、C、D三個系統發消息數據，如果B系統、C系統處理成功，D系統處理失敗。
如何保證消息數據處理的一致性？
1.3 各種MQ產品的比較
常見的MQ產品包括Kafka、ActiveMQ、RabbitMQ、RocketMQ。
轉存失敗重新上傳取消轉存失敗重新上傳取消
2. RocketMQ快速入門
RocketMQ是阿里巴巴2016年MQ中間件，使用Java語言開發，在阿里內部，RocketMQ承接了例如“雙11”等高并發場景的消息流轉，能夠處理萬億級別的消息。
2.1 準備工作
2.1.1 下載RocketMQ
RocketMQ最新版本：4.5.1
下載地址
2.2.2 環境要求
-
Linux64位系統
-
JDK1.8(64位)
-
源碼安裝需要安裝Maven 3.2.x
2.2 安裝RocketMQ
2.2.1 安裝步驟
本教程以二進制包方式安裝
- 解壓安裝包
- 進入安裝目錄
2.2.2 目錄介紹
- bin：啟動腳本，包括shell腳本和CMD腳本
- conf：實例配置文件 ，包括broker配置文件、logback配置文件等
- lib：依賴jar包，包括Netty、commons-lang、FastJSON等
2.3 啟動RocketMQ
- 啟動NameServer
# 1.啟動NameServer nohup sh bin/mqnamesrv & # 2.查看啟動日志 tail -f ~/logs/rocketmqlogs/namesrv.log
- 啟動Broker
# 1.啟動Broker nohup sh bin/mqbroker -n localhost:9876 & # 2.查看啟動日志 tail -f ~/logs/rocketmqlogs/broker.log
-
問題描述：
RocketMQ默認的虛擬機內存較大，啟動Broker如果因為內存不足失敗，需要編輯如下兩個配置文件，修改JVM內存大小
# 編輯runbroker.sh和runserver.sh修改默認JVM大小 vi runbroker.sh vi runserver.sh
- 參考設置：
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
2.4 測試RocketMQ
2.4.1 發送消息
# 1.設置環境變量 export NAMESRV_ADDR=localhost:9876 # 2.使用安裝包的Demo發送消息 sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
2.4.2 接收消息
# 1.設置環境變量 export NAMESRV_ADDR=localhost:9876 # 2.接收消息 sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
2.5 關閉RocketMQ
# 1.關閉NameServer sh bin/mqshutdown namesrv # 2.關閉Broker sh bin/mqshutdown broker
3. RocketMQ集群搭建
3.1 各角色介紹
- Producer：消息的發送者；舉例：發信者
- Consumer：消息接收者；舉例：收信者
- Broker：暫存和傳輸消息；舉例：郵局
- NameServer：管理Broker；舉例：各個郵局的管理機構
- Topic：區分消息的種類；一個發送者可以發送消息給一個或者多個Topic；一個消息的接收者可以訂閱一個或者多個Topic消息
- Message Queue：相當于是Topic的分區；用于并行發送和接收消息
轉存失敗重新上傳取消轉存失敗重新上傳取消
3.2 集群搭建方式
3.2.1 集群特點
-
NameServer是一個幾乎無狀態節點，可集群部署，節點之間無任何信息同步。
-
Broker部署相對復雜，Broker分為Master與Slave，一個Master可以對應多個Slave，但是一個Slave只能對應一個Master，Master與Slave的對應關系通過指定相同的BrokerName，不同的BrokerId來定義，BrokerId為0表示Master，非0表示Slave。Master也可以部署多個。每個Broker與NameServer集群中的所有節點建立長連接，定時注冊Topic信息到所有NameServer。
-
Producer與NameServer集群中的其中一個節點（隨機選擇）建立長連接，定期從NameServer取Topic路由信息，并向提供Topic服務的Master建立長連接，且定時向Master發送心跳。Producer完全無狀態，可集群部署。
-
Consumer與NameServer集群中的其中一個節點（隨機選擇）建立長連接，定期從NameServer取Topic路由信息，并向提供Topic服務的Master、Slave建立長連接，且定時向Master、Slave發送心跳。Consumer既可以從Master訂閱消息，也可以從Slave訂閱消息，訂閱規則由Broker配置決定。
3.2.3 集群模式
1）單Master模式
這種方式風險較大，一旦Broker重啟或者宕機時，會導致整個服務不可用。不建議線上環境使用,可以用于本地測試。
2）多Master模式
一個集群無Slave，全是Master，例如2個Master或者3個Master，這種模式的優缺點如下：
- 優點：配置簡單，單個Master宕機或重啟維護對應用無影響，在磁盤配置為RAID10時，即使機器宕機不可恢復情況下，由于RAID10磁盤非常可靠，消息也不會丟（異步刷盤丟失少量消息，同步刷盤一條不丟），性能最高；
- 缺點：單臺機器宕機期間，這臺機器上未被消費的消息在機器恢復之前不可訂閱，消息實時性會受到影響。
3）多Master多Slave模式（異步）
每個Master配置一個Slave，有多對Master-Slave，HA采用異步復制方式，主備有短暫消息延遲（毫秒級），這種模式的優缺點如下：
- 優點：即使磁盤損壞，消息丟失的非常少，且消息實時性不會受影響，同時Master宕機后，消費者仍然可以從Slave消費，而且此過程對應用透明，不需要人工干預，性能同多Master模式幾乎一樣；
- 缺點：Master宕機，磁盤損壞情況下會丟失少量消息。
4）多Master多Slave模式（同步）
每個Master配置一個Slave，有多對Master-Slave，HA采用同步雙寫方式，即只有主備都寫成功，才向應用返回成功，這種模式的優缺點如下：
- 優點：數據與服務都無單點故障，Master宕機情況下，消息無延遲，服務可用性與數據可用性都非常高；
- 缺點：性能比異步復制模式略低（大約低10%左右），發送單個消息的RT會略高，且目前版本在主節點宕機后，備機不能自動切換為主機。
3.3 雙主雙從集群搭建
3.3.1 總體架構
消息高可用采用2m-2s（同步雙寫）方式
轉存失敗重新上傳取消轉存失敗重新上傳取消
3.3.2 集群工作流程
- 啟動NameServer，NameServer起來后監聽端口，等待Broker、Producer、Consumer連上來，相當于一個路由控制中心。
- Broker啟動，跟所有的NameServer保持長連接，定時發送心跳包。心跳包中包含當前Broker信息(IP+端口等)以及存儲所有Topic信息。注冊成功后，NameServer集群中就有Topic跟Broker的映射關系。
- 收發消息前，先創建Topic，創建Topic時需要指定該Topic要存儲在哪些Broker上，也可以在發送消息時自動創建Topic。
- Producer發送消息，啟動時先跟NameServer集群中的其中一臺建立長連接，并從NameServer中獲取當前發送的Topic存在哪些Broker上，輪詢從隊列列表中選擇一個隊列，然后與隊列所在的Broker建立長連接從而向Broker發消息。
- Consumer跟Producer類似，跟其中一臺NameServer建立長連接，獲取當前訂閱Topic存在哪些Broker上，然后直接跟Broker建立連接通道，開始消費消息。
3.3.3 服務器環境
| 序號 | IP | 角色 | 架構模式 |
|---|---|---|---|
| 1 | 192.168.25.135 | nameserver、brokerserver | Master1、Slave2 |
| 2 | 192.168.25.138 | nameserver、brokerserver | Master2、Slave1 |
3.3.4 Host添加信息
vim /etc/hosts
配置如下:
# nameserver 192.168.25.135 rocketmq-nameserver1 192.168.25.138 rocketmq-nameserver2 # broker 192.168.25.135 rocketmq-master1 192.168.25.135 rocketmq-slave2 192.168.25.138 rocketmq-master2 192.168.25.138 rocketmq-slave1
配置完成后, 重啟網卡
systemctl restart network
3.3.5 防火墻配置
宿主機需要遠程訪問虛擬機的rocketmq服務和web服務，需要開放相關的端口號，簡單粗暴的方式是直接關閉防火墻
# 關閉防火墻 systemctl stop firewalld.service # 查看防火墻的狀態 firewall-cmd --state # 禁止firewall開機啟動 systemctl disable firewalld.service
或者為了安全，只開放特定的端口號，RocketMQ默認使用3個端口：9876 、10911 、11011 。如果防火墻沒有關閉的話，那么防火墻就必須開放這些端口：
nameserver?默認使用 9876 端口master?默認使用 10911 端口slave?默認使用11011 端口
執行以下命令：
# 開放name server默認端口 firewall-cmd --remove-port=9876/tcp --permanent # 開放master默認端口 firewall-cmd --remove-port=10911/tcp --permanent # 開放slave默認端口 (當前集群模式可不開啟) firewall-cmd --remove-port=11011/tcp --permanent # 重啟防火墻 firewall-cmd --reload
3.3.6 環境變量配置
vim /etc/profile
在profile文件的末尾加入如下命令
#set rocketmq ROCKETMQ_HOME=/usr/local/rocketmq/rocketmq-all-4.4.0-bin-release PATH=$PATH:$ROCKETMQ_HOME/bin export ROCKETMQ_HOME PATH
輸入:wq! 保存并退出， 并使得配置立刻生效：
source /etc/profile
3.3.7 創建消息存儲路徑
mkdir /usr/local/rocketmq/store mkdir /usr/local/rocketmq/store/commitlog mkdir /usr/local/rocketmq/store/consumequeue mkdir /usr/local/rocketmq/store/index
3.3.8 broker配置文件
1）master1
服務器：192.168.25.135
vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a.properties
修改配置如下：
#所屬集群名字 brokerClusterName=rocketmq-cluster #broker名字，注意此處不同的配置文件填寫的不一樣 brokerName=broker-a #0 表示 Master，>0 表示 Slave brokerId=0 #nameServer地址，分號分割 namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876 #在發送消息時，自動創建服務器不存在的topic，默認創建的隊列數 defaultTopicQueueNums=4 #是否允許 Broker 自動創建Topic，建議線下開啟，線上關閉 autoCreateTopicEnable=true #是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉 autoCreateSubscriptionGroup=true #Broker 對外服務的監聽端口 listenPort=10911 #刪除文件時間點，默認凌晨 4點 deleteWhen=04 #文件保留時間，默認 48 小時 fileReservedTime=120 #commitLog每個文件的大小默認1G mapedFileSizeCommitLog=1073741824 #ConsumeQueue每個文件默認存30W條，根據業務情況調整 mapedFileSizeConsumeQueue=300000 #destroyMapedFileIntervalForcibly=120000 #redeleteHangedFileInterval=120000 #檢測物理文件磁盤空間 diskMaxUsedSpaceRatio=88 #存儲路徑 storePathRootDir=/usr/local/rocketmq/store #commitLog 存儲路徑 storePathCommitLog=/usr/local/rocketmq/store/commitlog #消費隊列存儲路徑存儲路徑 storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue #消息索引存儲路徑 storePathIndex=/usr/local/rocketmq/store/index #checkpoint 文件存儲路徑 storeCheckpoint=/usr/local/rocketmq/store/checkpoint #abort 文件存儲路徑 abortFile=/usr/local/rocketmq/store/abort #限制的消息大小 maxMessageSize=65536 #flushCommitLogLeastPages=4 #flushConsumeQueueLeastPages=2 #flushCommitLogThoroughInterval=10000 #flushConsumeQueueThoroughInterval=60000 #Broker 的角色 #- ASYNC_MASTER 異步復制Master #- SYNC_MASTER 同步雙寫Master #- SLAVE brokerRole=SYNC_MASTER #刷盤方式 #- ASYNC_FLUSH 異步刷盤 #- SYNC_FLUSH 同步刷盤 flushDiskType=SYNC_FLUSH #checkTransactionMessageEnable=false #發消息線程池數量 #sendMessageThreadPoolNums=128 #拉消息線程池數量 #pullMessageThreadPoolNums=128
2）slave2
服務器：192.168.25.135
vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b-s.properties
修改配置如下：
#所屬集群名字 brokerClusterName=rocketmq-cluster #broker名字，注意此處不同的配置文件填寫的不一樣 brokerName=broker-b #0 表示 Master，>0 表示 Slave brokerId=1 #nameServer地址，分號分割 namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876 #在發送消息時，自動創建服務器不存在的topic，默認創建的隊列數 defaultTopicQueueNums=4 #是否允許 Broker 自動創建Topic，建議線下開啟，線上關閉 autoCreateTopicEnable=true #是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉 autoCreateSubscriptionGroup=true #Broker 對外服務的監聽端口 listenPort=11011 #刪除文件時間點，默認凌晨 4點 deleteWhen=04 #文件保留時間，默認 48 小時 fileReservedTime=120 #commitLog每個文件的大小默認1G mapedFileSizeCommitLog=1073741824 #ConsumeQueue每個文件默認存30W條，根據業務情況調整 mapedFileSizeConsumeQueue=300000 #destroyMapedFileIntervalForcibly=120000 #redeleteHangedFileInterval=120000 #檢測物理文件磁盤空間 diskMaxUsedSpaceRatio=88 #存儲路徑 storePathRootDir=/usr/local/rocketmq/store #commitLog 存儲路徑 storePathCommitLog=/usr/local/rocketmq/store/commitlog #消費隊列存儲路徑存儲路徑 storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue #消息索引存儲路徑 storePathIndex=/usr/local/rocketmq/store/index #checkpoint 文件存儲路徑 storeCheckpoint=/usr/local/rocketmq/store/checkpoint #abort 文件存儲路徑 abortFile=/usr/local/rocketmq/store/abort #限制的消息大小 maxMessageSize=65536 #flushCommitLogLeastPages=4 #flushConsumeQueueLeastPages=2 #flushCommitLogThoroughInterval=10000 #flushConsumeQueueThoroughInterval=60000 #Broker 的角色 #- ASYNC_MASTER 異步復制Master #- SYNC_MASTER 同步雙寫Master #- SLAVE brokerRole=SLAVE #刷盤方式 #- ASYNC_FLUSH 異步刷盤 #- SYNC_FLUSH 同步刷盤 flushDiskType=ASYNC_FLUSH #checkTransactionMessageEnable=false #發消息線程池數量 #sendMessageThreadPoolNums=128 #拉消息線程池數量 #pullMessageThreadPoolNums=128
3）master2
服務器：192.168.25.138
vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b.properties
修改配置如下：
#所屬集群名字 brokerClusterName=rocketmq-cluster #broker名字，注意此處不同的配置文件填寫的不一樣 brokerName=broker-b #0 表示 Master，>0 表示 Slave brokerId=0 #nameServer地址，分號分割 namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876 #在發送消息時，自動創建服務器不存在的topic，默認創建的隊列數 defaultTopicQueueNums=4 #是否允許 Broker 自動創建Topic，建議線下開啟，線上關閉 autoCreateTopicEnable=true #是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉 autoCreateSubscriptionGroup=true #Broker 對外服務的監聽端口 listenPort=10911 #刪除文件時間點，默認凌晨 4點 deleteWhen=04 #文件保留時間，默認 48 小時 fileReservedTime=120 #commitLog每個文件的大小默認1G mapedFileSizeCommitLog=1073741824 #ConsumeQueue每個文件默認存30W條，根據業務情況調整 mapedFileSizeConsumeQueue=300000 #destroyMapedFileIntervalForcibly=120000 #redeleteHangedFileInterval=120000 #檢測物理文件磁盤空間 diskMaxUsedSpaceRatio=88 #存儲路徑 storePathRootDir=/usr/local/rocketmq/store #commitLog 存儲路徑 storePathCommitLog=/usr/local/rocketmq/store/commitlog #消費隊列存儲路徑存儲路徑 storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue #消息索引存儲路徑 storePathIndex=/usr/local/rocketmq/store/index #checkpoint 文件存儲路徑 storeCheckpoint=/usr/local/rocketmq/store/checkpoint #abort 文件存儲路徑 abortFile=/usr/local/rocketmq/store/abort #限制的消息大小 maxMessageSize=65536 #flushCommitLogLeastPages=4 #flushConsumeQueueLeastPages=2 #flushCommitLogThoroughInterval=10000 #flushConsumeQueueThoroughInterval=60000 #Broker 的角色 #- ASYNC_MASTER 異步復制Master #- SYNC_MASTER 同步雙寫Master #- SLAVE brokerRole=SYNC_MASTER #刷盤方式 #- ASYNC_FLUSH 異步刷盤 #- SYNC_FLUSH 同步刷盤 flushDiskType=SYNC_FLUSH #checkTransactionMessageEnable=false #發消息線程池數量 #sendMessageThreadPoolNums=128 #拉消息線程池數量 #pullMessageThreadPoolNums=128
4）slave1
服務器：192.168.25.138
vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a-s.properties
修改配置如下：
#所屬集群名字 brokerClusterName=rocketmq-cluster #broker名字，注意此處不同的配置文件填寫的不一樣 brokerName=broker-a #0 表示 Master，>0 表示 Slave brokerId=1 #nameServer地址，分號分割 namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876 #在發送消息時，自動創建服務器不存在的topic，默認創建的隊列數 defaultTopicQueueNums=4 #是否允許 Broker 自動創建Topic，建議線下開啟，線上關閉 autoCreateTopicEnable=true #是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉 autoCreateSubscriptionGroup=true #Broker 對外服務的監聽端口 listenPort=11011 #刪除文件時間點，默認凌晨 4點 deleteWhen=04 #文件保留時間，默認 48 小時 fileReservedTime=120 #commitLog每個文件的大小默認1G mapedFileSizeCommitLog=1073741824 #ConsumeQueue每個文件默認存30W條，根據業務情況調整 mapedFileSizeConsumeQueue=300000 #destroyMapedFileIntervalForcibly=120000 #redeleteHangedFileInterval=120000 #檢測物理文件磁盤空間 diskMaxUsedSpaceRatio=88 #存儲路徑 storePathRootDir=/usr/local/rocketmq/store #commitLog 存儲路徑 storePathCommitLog=/usr/local/rocketmq/store/commitlog #消費隊列存儲路徑存儲路徑 storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue #消息索引存儲路徑 storePathIndex=/usr/local/rocketmq/store/index #checkpoint 文件存儲路徑 storeCheckpoint=/usr/local/rocketmq/store/checkpoint #abort 文件存儲路徑 abortFile=/usr/local/rocketmq/store/abort #限制的消息大小 maxMessageSize=65536 #flushCommitLogLeastPages=4 #flushConsumeQueueLeastPages=2 #flushCommitLogThoroughInterval=10000 #flushConsumeQueueThoroughInterval=60000 #Broker 的角色 #- ASYNC_MASTER 異步復制Master #- SYNC_MASTER 同步雙寫Master #- SLAVE brokerRole=SLAVE #刷盤方式 #- ASYNC_FLUSH 異步刷盤 #- SYNC_FLUSH 同步刷盤 flushDiskType=ASYNC_FLUSH #checkTransactionMessageEnable=false #發消息線程池數量 #sendMessageThreadPoolNums=128 #拉消息線程池數量 #pullMessageThreadPoolNums=128
3.3.9 修改啟動腳本文件
1）runbroker.sh
vi /usr/local/rocketmq/bin/runbroker.sh
需要根據內存大小進行適當的對JVM參數進行調整：
#===================================================
# 開發環境配置 JVM Configuration
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"
####2）runserver.sh
vim /usr/local/rocketmq/bin/runserver.sh
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
3.3.10 服務啟動
1）啟動NameServe集群
分別在192.168.25.135和192.168.25.138啟動NameServer
cd /usr/local/rocketmq/bin nohup sh mqnamesrv &
2）啟動Broker集群
- 在192.168.25.135上啟動master1和slave2
master1：
cd /usr/local/rocketmq/bin nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-syncbroker-a.properties &
slave2：
cd /usr/local/rocketmq/bin nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b-s.properties &
- 在192.168.25.138上啟動master2和slave2
master2
cd /usr/local/rocketmq/bin nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b.properties &
slave1
cd /usr/local/rocketmq/bin nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-a-s.properties &
3.3.11 查看進程狀態
啟動后通過JPS查看啟動進程
轉存失敗重新上傳取消轉存失敗重新上傳取消
3.3.12 查看日志
# 查看nameServer日志 tail -500f ~/logs/rocketmqlogs/namesrv.log # 查看broker日志 tail -500f ~/logs/rocketmqlogs/broker.log
3.4 mqadmin管理工具
3.4.1 使用方式
進入RocketMQ安裝位置，在bin目錄下執行./mqadmin {command} {args}
###3.4.2 命令介紹
####1）Topic相關
| 名稱 | 含義 | 命令選項 | 說明 |
| updateTopic | 創建更新Topic配置 | -b | Broker 地址，表示 topic 所在 Broker，只支持單臺Broker，地址為ip:port |
| -c | cluster 名稱，表示 topic 所在集群（集群可通過 clusterList 查詢） | ||
| -h- | 打印幫助 | ||
| -n | NameServer服務地址，格式 ip:port | ||
| -p | 指定新topic的讀寫權限( W=2|R=4|WR=6 ) | ||
| -r | 可讀隊列數（默認為 8） | ||
| -w | 可寫隊列數（默認為 8） | ||
| -t | topic 名稱（名稱只能使用字符 ^[a-zA-Z0-9_-]+$ ） | ||
| deleteTopic | 刪除Topic | -c | cluster 名稱，表示刪除某集群下的某個 topic （集群 可通過 clusterList 查詢） |
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| -t | topic 名稱（名稱只能使用字符 ^[a-zA-Z0-9_-]+$ ） | ||
| topicList | 查看 Topic 列表信息 | -h | 打印幫助 |
| -c | 不配置-c只返回topic列表，增加-c返回clusterName, topic, consumerGroup信息，即topic的所屬集群和訂閱關系，沒有參數 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| topicRoute | 查看 Topic 路由信息 | -t | topic 名稱 |
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| topicStatus | 查看 Topic 消息隊列offset | -t | topic 名稱 |
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| topicClusterList | 查看 Topic 所在集群列表 | -t | topic 名稱 |
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| updateTopicPerm | 更新 Topic 讀寫權限 | -t | topic 名稱 |
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| -b | Broker 地址，表示 topic 所在 Broker，只支持單臺Broker，地址為ip:port | ||
| -p | 指定新 topic 的讀寫權限( W=2|R=4|WR=6 ) | ||
| -c | cluster 名稱，表示 topic 所在集群（集群可通過 clusterList 查詢），-b優先，如果沒有-b，則對集群中所有Broker執行命令 | ||
| updateOrderConf | 從NameServer上創建、刪除、獲取特定命名空間的kv配置，目前還未啟用 | -h | 打印幫助 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -t | topic，鍵 | ||
| -v | orderConf，值 | ||
| -m | method，可選get、put、delete | ||
| allocateMQ | 以平均負載算法計算消費者列表負載消息隊列的負載結果 | -t | topic 名稱 |
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| -i | ipList，用逗號分隔，計算這些ip去負載Topic的消息隊列 | ||
| statsAll | 打印Topic訂閱關系、TPS、積累量、24h讀寫總量等信息 | -h | 打印幫助 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -a | 是否只打印活躍topic | ||
| -t | 指定topic |
####2）集群相關
| 名稱 | 含義 | 命令選項 | 說明 |
| clusterList | 查看集群信息，集群、BrokerName、BrokerId、TPS等信息 | -m | 打印更多信息 (增加打印出如下信息 #InTotalYest, #OutTotalYest, #InTotalToday ,#OutTotalToday) |
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| -i | 打印間隔，單位秒 | ||
| clusterRT | 發送消息檢測集群各Broker RT。消息發往${BrokerName} Topic。 | -a | amount，每次探測的總數，RT = 總時間 / amount |
| -s | 消息大小，單位B | ||
| -c | 探測哪個集群 | ||
| -p | 是否打印格式化日志，以|分割，默認不打印 | ||
| -h | 打印幫助 | ||
| -m | 所屬機房，打印使用 | ||
| -i | 發送間隔，單位秒 | ||
| -n | NameServer 服務地址，格式 ip:port |
####3）Broker相關
| 名稱 | 含義 | 命令選項 | 說明 |
| updateBrokerConfig | 更新 Broker 配置文件，會修改Broker.conf | -b | Broker 地址，格式為ip:port |
| -c | cluster 名稱 | ||
| -k | key 值 | ||
| -v | value 值 | ||
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| brokerStatus | 查看 Broker 統計信息、運行狀態（你想要的信息幾乎都在里面） | -b | Broker 地址，地址為ip:port |
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| brokerConsumeStats | Broker中各個消費者的消費情況，按Message Queue維度返回Consume Offset，Broker Offset，Diff，TImestamp等信息 | -b | Broker 地址，地址為ip:port |
| -t | 請求超時時間 | ||
| -l | diff閾值，超過閾值才打印 | ||
| -o | 是否為順序topic，一般為false | ||
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| getBrokerConfig | 獲取Broker配置 | -b | Broker 地址，地址為ip:port |
| -n | NameServer 服務地址，格式 ip:port | ||
| wipeWritePerm | 從NameServer上清除 Broker寫權限 | -b | Broker 地址，地址為ip:port |
| -n | NameServer 服務地址，格式 ip:port | ||
| -h | 打印幫助 | ||
| cleanExpiredCQ | 清理Broker上過期的Consume Queue，如果手動減少對列數可能產生過期隊列 | -n | NameServer 服務地址，格式 ip:port |
| -h | 打印幫助 | ||
| -b | Broker 地址，地址為ip:port | ||
| -c | 集群名稱 | ||
| cleanUnusedTopic | 清理Broker上不使用的Topic，從內存中釋放Topic的Consume Queue，如果手動刪除Topic會產生不使用的Topic | -n | NameServer 服務地址，格式 ip:port |
| -h | 打印幫助 | ||
| -b | Broker 地址，地址為ip:port | ||
| -c | 集群名稱 | ||
| sendMsgStatus | 向Broker發消息，返回發送狀態和RT | -n | NameServer 服務地址，格式 ip:port |
| -h | 打印幫助 | ||
| -b | BrokerName，注意不同于Broker地址 | ||
| -s | 消息大小，單位B | ||
| -c | 發送次數 |
####4）消息相關
| 名稱 | 含義 | 命令選項 | 說明 |
| queryMsgById | 根據offsetMsgId查詢msg，如果使用開源控制臺，應使用offsetMsgId，此命令還有其他參數，具體作用請閱讀QueryMsgByIdSubCommand。 | -i | msgId |
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| queryMsgByKey | 根據消息 Key 查詢消息 | -k | msgKey |
| -t | Topic 名稱 | ||
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| queryMsgByOffset | 根據 Offset 查詢消息 | -b | Broker 名稱，（這里需要注意 填寫的是 Broker 的名稱，不是 Broker 的地址，Broker 名稱可以在 clusterList 查到） |
| -i | query 隊列 id | ||
| -o | offset 值 | ||
| -t | topic 名稱 | ||
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| queryMsgByUniqueKey | 根據msgId查詢，msgId不同于offsetMsgId，區別詳見常見運維問題。-g，-d配合使用，查到消息后嘗試讓特定的消費者消費消息并返回消費結果 | -h | 打印幫助 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -i | uniqe msg id | ||
| -g | consumerGroup | ||
| -d | clientId | ||
| -t | topic名稱 | ||
| checkMsgSendRT | 檢測向topic發消息的RT，功能類似clusterRT | -h | 打印幫助 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -t | topic名稱 | ||
| -a | 探測次數 | ||
| -s | 消息大小 | ||
| sendMessage | 發送一條消息，可以根據配置發往特定Message Queue，或普通發送。 | -h | 打印幫助 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -t | topic名稱 | ||
| -p | body，消息體 | ||
| -k | keys | ||
| -c | tags | ||
| -b | BrokerName | ||
| -i | queueId | ||
| consumeMessage | 消費消息。可以根據offset、開始&結束時間戳、消息隊列消費消息，配置不同執行不同消費邏輯，詳見ConsumeMessageCommand。 | -h | 打印幫助 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -t | topic名稱 | ||
| -b | BrokerName | ||
| -o | 從offset開始消費 | ||
| -i | queueId | ||
| -g | 消費者分組 | ||
| -s | 開始時間戳，格式詳見-h | ||
| -d | 結束時間戳 | ||
| -c | 消費多少條消息 | ||
| printMsg | 從Broker消費消息并打印，可選時間段 | -h | 打印幫助 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -t | topic名稱 | ||
| -c | 字符集，例如UTF-8 | ||
| -s | subExpress，過濾表達式 | ||
| -b | 開始時間戳，格式參見-h | ||
| -e | 結束時間戳 | ||
| -d | 是否打印消息體 | ||
| printMsgByQueue | 類似printMsg，但指定Message Queue | -h | 打印幫助 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -t | topic名稱 | ||
| -i | queueId | ||
| -a | BrokerName | ||
| -c | 字符集，例如UTF-8 | ||
| -s | subExpress，過濾表達式 | ||
| -b | 開始時間戳，格式參見-h | ||
| -e | 結束時間戳 | ||
| -p | 是否打印消息 | ||
| -d | 是否打印消息體 | ||
| -f | 是否統計tag數量并打印 | ||
| resetOffsetByTime | 按時間戳重置offset，Broker和consumer都會重置 | -h | 打印幫助 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -g | 消費者分組 | ||
| -t | topic名稱 | ||
| -s | 重置為此時間戳對應的offset | ||
| -f | 是否強制重置，如果false，只支持回溯offset，如果true，不管時間戳對應offset與consumeOffset關系 | ||
| -c | 是否重置c++客戶端offset |
5）消費者、消費組相關
| 名稱 | 含義 | 命令選項 | 說明 |
| consumerProgress | 查看訂閱組消費狀態，可以查看具體的client IP的消息積累量 | -g | 消費者所屬組名 |
| -s | 是否打印client IP | ||
| -h | 打印幫助 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| consumerStatus | 查看消費者狀態，包括同一個分組中是否都是相同的訂閱，分析Process Queue是否堆積，返回消費者jstack結果，內容較多，使用者參見ConsumerStatusSubCommand | -h | 打印幫助 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -g | consumer group | ||
| -i | clientId | ||
| -s | 是否執行jstack | ||
| getConsumerStatus | 獲取 Consumer 消費進度 | -g | 消費者所屬組名 |
| -t | 查詢主題 | ||
| -i | Consumer 客戶端 ip | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| -h | 打印幫助 | ||
| updateSubGroup | 更新或創建訂閱關系 | -n | NameServer 服務地址，格式 ip:port |
| -h | 打印幫助 | ||
| -b | Broker地址 | ||
| -c | 集群名稱 | ||
| -g | 消費者分組名稱 | ||
| -s | 分組是否允許消費 | ||
| -m | 是否從最小offset開始消費 | ||
| -d | 是否是廣播模式 | ||
| -q | 重試隊列數量 | ||
| -r | 最大重試次數 | ||
| -i | 當slaveReadEnable開啟時有效，且還未達到從slave消費時建議從哪個BrokerId消費，可以配置備機id，主動從備機消費 | ||
| -w | 如果Broker建議從slave消費，配置決定從哪個slave消費，配置BrokerId，例如1 | ||
| -a | 當消費者數量變化時是否通知其他消費者負載均衡 | ||
| deleteSubGroup | 從Broker刪除訂閱關系 | -n | NameServer 服務地址，格式 ip:port |
| -h | 打印幫助 | ||
| -b | Broker地址 | ||
| -c | 集群名稱 | ||
| -g | 消費者分組名稱 | ||
| cloneGroupOffset | 在目標群組中使用源群組的offset | -n | NameServer 服務地址，格式 ip:port |
| -h | 打印幫助 | ||
| -s | 源消費者組 | ||
| -d | 目標消費者組 | ||
| -t | topic名稱 | ||
| -o | 暫未使用 |
6）連接相關
| 名稱 | 含義 | 命令選項 | 說明 |
| consumerConnec tion | 查詢 Consumer 的網絡連接 | -g | 消費者所屬組名 |
| -n | NameServer 服務地址，格式 ip:port | ||
| -h | 打印幫助 | ||
| producerConnec tion | 查詢 Producer 的網絡連接 | -g | 生產者所屬組名 |
| -t | 主題名稱 | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| -h | 打印幫助 |
7）NameServer相關
| 名稱 | 含義 | 命令選項 | 說明 |
| updateKvConfig | 更新NameServer的kv配置，目前還未使用 | -s | 命名空間 |
| -k | key | ||
| -v | value | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| -h | 打印幫助 | ||
| deleteKvConfig | 刪除NameServer的kv配置 | -s | 命名空間 |
| -k | key | ||
| -n | NameServer 服務地址，格式 ip:port | ||
| -h | 打印幫助 | ||
| getNamesrvConfig | 獲取NameServer配置 | -n | NameServer 服務地址，格式 ip:port |
| -h | 打印幫助 | ||
| updateNamesrvConfig | 修改NameServer配置 | -n | NameServer 服務地址，格式 ip:port |
| -h | 打印幫助 | ||
| -k | key | ||
| -v | value |
8）其他
| 名稱 | 含義 | 命令選項 | 說明 |
| startMonitoring | 開啟監控進程，監控消息誤刪、重試隊列消息數等 | -n | NameServer 服務地址，格式 ip:port |
| -h | 打印幫助 |
3.4.3 注意事項
- 幾乎所有命令都需要配置-n表示NameServer地址，格式為ip:port
- 幾乎所有命令都可以通過-h獲取幫助
- 如果既有Broker地址（-b）配置項又有clusterName（-c）配置項，則優先以Broker地址執行命令；如果不配置Broker地址，則對集群中所有主機執行命令
3.5 集群監控平臺搭建
3.5.1 概述
RocketMQ有一個對其擴展的開源項目incubator-rocketmq-externals，這個項目中有一個子模塊叫rocketmq-console，這個便是管理控制臺項目了，先將incubator-rocketmq-externals拉到本地，因為我們需要自己對rocketmq-console進行編譯打包運行。
轉存失敗重新上傳取消轉存失敗重新上傳取消
3.5.2 下載并編譯打包
git clone https://github.com/apache/rocketmq-externals cd rocketmq-console mvn clean package -Dmaven.test.skip=true
注意：打包前在rocketmq-console中配置namesrv集群地址：
rocketmq.config.namesrvAddr=192.168.25.135:9876;192.168.25.138:9876
啟動rocketmq-console：
java -jar rocketmq-console-ng-1.0.0.jar
啟動成功后，我們就可以通過瀏覽器訪問http://localhost:8080進入控制臺界面了，如下圖：
轉存失敗重新上傳取消
集群狀態：
轉存失敗重新上傳取消
4. 消息發送樣例
- 導入MQ客戶端依賴
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.4.0</version> </dependency>
- 消息發送者步驟分析r
1.創建消息生產者producer，并制定生產者組名 2.指定Nameserver地址 3.啟動producer 4.創建消息對象，指定主題Topic、Tag和消息體 5.發送消息 6.關閉生產者producer
- 消息消費者步驟分析
1.創建消費者Consumer，制定消費者組名 2.指定Nameserver地址 3.訂閱主題Topic和Tag 4.設置回調函數，處理消息 5.啟動消費者consumer
4.1 基本樣例
4.1.1 消息發送
1）發送同步消息
這種可靠性同步地發送方式使用的比較廣泛，比如：重要的消息通知，短信通知。
public class SyncProducer {public static void main(String[] args) throws Exception {// 實例化消息生產者ProducerDefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");// 設置NameServer的地址producer.setNamesrvAddr("localhost:9876");// 啟動Producer實例producer.start();for (int i = 0; i < 100; i++) {// 創建消息，并指定Topic，Tag和消息體Message msg = new Message("TopicTest" /* Topic */,"TagA" /* Tag */,("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */);// 發送消息到一個BrokerSendResult sendResult = producer.send(msg);// 通過sendResult返回消息是否成功送達System.out.printf("%s%n", sendResult);}// 如果不再發送消息，關閉Producer實例。producer.shutdown();}
}
2）發送異步消息
異步消息通常用在對響應時間敏感的業務場景，即發送端不能容忍長時間地等待Broker的響應。
public class AsyncProducer {public static void main(String[] args) throws Exception {// 實例化消息生產者ProducerDefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");// 設置NameServer的地址producer.setNamesrvAddr("localhost:9876");// 啟動Producer實例producer.start();producer.setRetryTimesWhenSendAsyncFailed(0);for (int i = 0; i < 100; i++) {final int index = i;// 創建消息，并指定Topic，Tag和消息體Message msg = new Message("TopicTest","TagA","OrderID188","Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));// SendCallback接收異步返回結果的回調producer.send(msg, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.printf("%-10d OK %s %n", index,sendResult.getMsgId());}@Overridepublic void onException(Throwable e) {System.out.printf("%-10d Exception %s %n", index, e);e.printStackTrace();}});}// 如果不再發送消息，關閉Producer實例。producer.shutdown();}
}
3）單向發送消息
這種方式主要用在不特別關心發送結果的場景，例如日志發送。
public class OnewayProducer {public static void main(String[] args) throws Exception{// 實例化消息生產者ProducerDefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");// 設置NameServer的地址producer.setNamesrvAddr("localhost:9876");// 啟動Producer實例producer.start();for (int i = 0; i < 100; i++) {// 創建消息，并指定Topic，Tag和消息體Message msg = new Message("TopicTest" /* Topic */,"TagA" /* Tag */,("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */);// 發送單向消息，沒有任何返回結果producer.sendOneway(msg);}// 如果不再發送消息，關閉Producer實例。producer.shutdown();}
}
4.1.2 消費消息
1）負載均衡模式
消費者采用負載均衡方式消費消息，多個消費者共同消費隊列消息，每個消費者處理的消息不同
public static void main(String[] args) throws Exception {// 實例化消息生產者,指定組名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");// 指定Namesrv地址信息.consumer.setNamesrvAddr("localhost:9876");// 訂閱Topicconsumer.subscribe("Test", "*");//負載均衡模式消費consumer.setMessageModel(MessageModel.CLUSTERING);// 注冊回調函數，處理消息consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//啟動消息者consumer.start();System.out.printf("Consumer Started.%n");
}
2）廣播模式
消費者采用廣播的方式消費消息，每個消費者消費的消息都是相同的
public static void main(String[] args) throws Exception {// 實例化消息生產者,指定組名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");// 指定Namesrv地址信息.consumer.setNamesrvAddr("localhost:9876");// 訂閱Topicconsumer.subscribe("Test", "*");//廣播模式消費consumer.setMessageModel(MessageModel.BROADCASTING);// 注冊回調函數，處理消息consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//啟動消息者consumer.start();System.out.printf("Consumer Started.%n");
}
4.2 順序消息
消息有序指的是可以按照消息的發送順序來消費(FIFO)。RocketMQ可以嚴格的保證消息有序，可以分為分區有序或者全局有序。
順序消費的原理解析，在默認的情況下消息發送會采取Round Robin輪詢方式把消息發送到不同的queue(分區隊列)；而消費消息的時候從多個queue上拉取消息，這種情況發送和消費是不能保證順序。但是如果控制發送的順序消息只依次發送到同一個queue中，消費的時候只從這個queue上依次拉取，則就保證了順序。當發送和消費參與的queue只有一個，則是全局有序；如果多個queue參與，則為分區有序，即相對每個queue，消息都是有序的。
下面用訂單進行分區有序的示例。一個訂單的順序流程是：創建、付款、推送、完成。訂單號相同的消息會被先后發送到同一個隊列中，消費時，同一個OrderId獲取到的肯定是同一個隊列。
4.2.1 順序消息生產
/**
* Producer，發送順序消息
*/
public class Producer {public static void main(String[] args) throws Exception {DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();String[] tags = new String[]{"TagA", "TagC", "TagD"};// 訂單列表List<OrderStep> orderList = new Producer().buildOrders();Date date = new Date();SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");String dateStr = sdf.format(date);for (int i = 0; i < 10; i++) {// 加個時間前綴String body = dateStr + " Hello RocketMQ " + orderList.get(i);Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i, body.getBytes());SendResult sendResult = producer.send(msg, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {Long id = (Long) arg; //根據訂單id選擇發送queuelong index = id % mqs.size();return mqs.get((int) index);}}, orderList.get(i).getOrderId());//訂單idSystem.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",sendResult.getSendStatus(),sendResult.getMessageQueue().getQueueId(),body));}producer.shutdown();}/*** 訂單的步驟*/private static class OrderStep {private long orderId;private String desc;public long getOrderId() {return orderId;}public void setOrderId(long orderId) {this.orderId = orderId;}public String getDesc() {return desc;}public void setDesc(String desc) {this.desc = desc;}@Overridepublic String toString() {return "OrderStep{" +"orderId=" + orderId +", desc='" + desc + '\'' +'}';}}/*** 生成模擬訂單數據*/private List<OrderStep> buildOrders() {List<OrderStep> orderList = new ArrayList<OrderStep>();OrderStep orderDemo = new OrderStep();orderDemo.setOrderId(15103111039L);orderDemo.setDesc("創建");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(15103111065L);orderDemo.setDesc("創建");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(15103111039L);orderDemo.setDesc("付款");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(15103117235L);orderDemo.setDesc("創建");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(15103111065L);orderDemo.setDesc("付款");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(15103117235L);orderDemo.setDesc("付款");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(15103111065L);orderDemo.setDesc("完成");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(15103111039L);orderDemo.setDesc("推送");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(15103117235L);orderDemo.setDesc("完成");orderList.add(orderDemo);orderDemo = new OrderStep();orderDemo.setOrderId(15103111039L);orderDemo.setDesc("完成");orderList.add(orderDemo);return orderList;}
}
4.2.2 順序消費消息
/**
* 順序消息消費，帶事務方式（應用可控制Offset什么時候提交）
*/
public class ConsumerInOrder {public static void main(String[] args) throws Exception {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");consumer.setNamesrvAddr("127.0.0.1:9876");/*** 設置Consumer第一次啟動是從隊列頭部開始消費還是隊列尾部開始消費<br>* 如果非第一次啟動，那么按照上次消費的位置繼續消費*/consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.subscribe("TopicTest", "TagA || TagC || TagD");consumer.registerMessageListener(new MessageListenerOrderly() {Random random = new Random();@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {context.setAutoCommit(true);for (MessageExt msg : msgs) {// 可以看到每個queue有唯一的consume線程來消費, 訂單對每個queue(分區)有序System.out.println("consumeThread=" + Thread.currentThread().getName() + "queueId=" + msg.getQueueId() + ", content:" + new String(msg.getBody()));}try {//模擬業務邏輯處理中...TimeUnit.SECONDS.sleep(random.nextInt(10));} catch (Exception e) {e.printStackTrace();}return ConsumeOrderlyStatus.SUCCESS;}});consumer.start();System.out.println("Consumer Started.");}
}
4.3 延時消息
比如電商里，提交了一個訂單就可以發送一個延時消息，1h后去檢查這個訂單的狀態，如果還是未付款就取消訂單釋放庫存。
4.3.1 啟動消息消費者
public class ScheduledMessageConsumer {public static void main(String[] args) throws Exception {// 實例化消費者DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");// 訂閱Topicsconsumer.subscribe("TestTopic", "*");// 注冊消息監聽者consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {for (MessageExt message : messages) {// Print approximate delay time periodSystem.out.println("Receive message[msgId=" + message.getMsgId() + "] " + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later");}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 啟動消費者consumer.start();}
}
4.3.2 發送延時消息
public class ScheduledMessageProducer {public static void main(String[] args) throws Exception {// 實例化一個生產者來產生延時消息DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");// 啟動生產者producer.start();int totalMessagesToSend = 100;for (int i = 0; i < totalMessagesToSend; i++) {Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());// 設置延時等級3,這個消息將在10s之后發送(現在只支持固定的幾個時間,詳看delayTimeLevel)message.setDelayTimeLevel(3);// 發送消息producer.send(message);}// 關閉生產者producer.shutdown();}
}
###4.3.3 驗證
您將會看到消息的消費比存儲時間晚10秒
4.3.4 使用限制
// org/apache/rocketmq/store/config/MessageStoreConfig.java private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
現在RocketMq并不支持任意時間的延時，需要設置幾個固定的延時等級，從1s到2h分別對應著等級1到18
4.4 批量消息
批量發送消息能顯著提高傳遞小消息的性能。限制是這些批量消息應該有相同的topic，相同的waitStoreMsgOK，而且不能是延時消息。此外，這一批消息的總大小不應超過4MB。
4.4.1 發送批量消息
如果您每次只發送不超過4MB的消息，則很容易使用批處理，樣例如下：
String topic = "BatchTest";
List<Message> messages = new ArrayList<>();
messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
try {producer.send(messages);
} catch (Exception e) {e.printStackTrace();//處理error
}
如果消息的總長度可能大于4MB時，這時候最好把消息進行分割
public class ListSplitter implements Iterator<List<Message>> {private final int SIZE_LIMIT = 1024 * 1024 * 4;private final List<Message> messages;private int currIndex;public ListSplitter(List<Message> messages) {this.messages = messages;}@Override public boolean hasNext() {return currIndex < messages.size();}@Override public List<Message> next() {int nextIndex = currIndex;int totalSize = 0;for (; nextIndex < messages.size(); nextIndex++) {Message message = messages.get(nextIndex);int tmpSize = message.getTopic().length() + message.getBody().length;Map<String, String> properties = message.getProperties();for (Map.Entry<String, String> entry : properties.entrySet()) {tmpSize += entry.getKey().length() + entry.getValue().length();}tmpSize = tmpSize + 20; // 增加日志的開銷20字節if (tmpSize > SIZE_LIMIT) {//單個消息超過了最大的限制//忽略,否則會阻塞分裂的進程if (nextIndex - currIndex == 0) {//假如下一個子列表沒有元素,則添加這個子列表然后退出循環,否則只是退出循環nextIndex++;}break;}if (tmpSize + totalSize > SIZE_LIMIT) {break;} else {totalSize += tmpSize;}}List<Message> subList = messages.subList(currIndex, nextIndex);currIndex = nextIndex;return subList;}
}
//把大的消息分裂成若干個小的消息
ListSplitter splitter = new ListSplitter(messages);
while (splitter.hasNext()) {try {List<Message> listItem = splitter.next();producer.send(listItem);} catch (Exception e) {e.printStackTrace();//處理error}
}
4.5 過濾消息
在大多數情況下，TAG是一個簡單而有用的設計，其可以來選擇您想要的消息。例如：
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");
消費者將接收包含TAGA或TAGB或TAGC的消息。但是限制是一個消息只能有一個標簽，這對于復雜的場景可能不起作用。在這種情況下，可以使用SQL表達式篩選消息。SQL特性可以通過發送消息時的屬性來進行計算。在RocketMQ定義的語法下，可以實現一些簡單的邏輯。下面是一個例子：
------------
| message |
|----------| a > 5 AND b = 'abc'
| a = 10 | --------------------> Gotten
| b = 'abc'|
| c = true |
------------
------------
| message |
|----------| a > 5 AND b = 'abc'
| a = 1 | --------------------> Missed
| b = 'abc'|
| c = true |
------------
4.5.1 SQL基本語法
RocketMQ只定義了一些基本語法來支持這個特性。你也可以很容易地擴展它。
- 數值比較，比如：>，>=，<，<=，BETWEEN，=；
- 字符比較，比如：=，<>，IN；
- IS NULL?或者?IS NOT NULL；
- 邏輯符號?AND，OR，NOT；
常量支持類型為：
- 數值，比如：123，3.1415；
- 字符，比如：'abc'，必須用單引號包裹起來；
- NULL，特殊的常量
- 布爾值，TRUE?或?FALSE
只有使用push模式的消費者才能用使用SQL92標準的sql語句，接口如下：
public void subscribe(finalString topic, final MessageSelector messageSelector)
4.5.2 消息生產者
發送消息時，你能通過putUserProperty來設置消息的屬性
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
producer.start();
Message msg = new Message("TopicTest",tag,("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
);
// 設置一些屬性
msg.putUserProperty("a", String.valueOf(i));
SendResult sendResult = producer.send(msg);producer.shutdown();
4.5.3 消息消費者
用MessageSelector.bySql來使用sql篩選消息
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
// 只有訂閱的消息有這個屬性a, a >=0 and a <= 3
consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");
consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
});
consumer.start();
4.6 事務消息
###4.6.1 流程分析
轉存失敗重新上傳取消轉存失敗重新上傳取消
上圖說明了事務消息的大致方案，其中分為兩個流程：正常事務消息的發送及提交、事務消息的補償流程。
####1）事務消息發送及提交
(1) 發送消息（half消息）。
(2) 服務端響應消息寫入結果。
(3) 根據發送結果執行本地事務（如果寫入失敗，此時half消息對業務不可見，本地邏輯不執行）。
(4) 根據本地事務狀態執行Commit或者Rollback（Commit操作生成消息索引，消息對消費者可見）
2）事務補償
(1) 對沒有Commit/Rollback的事務消息（pending狀態的消息），從服務端發起一次“回查”
(2) Producer收到回查消息，檢查回查消息對應的本地事務的狀態
(3) 根據本地事務狀態，重新Commit或者Rollback
其中，補償階段用于解決消息Commit或者Rollback發生超時或者失敗的情況。
3）事務消息狀態
事務消息共有三種狀態，提交狀態、回滾狀態、中間狀態：
- TransactionStatus.CommitTransaction: 提交事務，它允許消費者消費此消息。
- TransactionStatus.RollbackTransaction: 回滾事務，它代表該消息將被刪除，不允許被消費。
- TransactionStatus.Unknown: 中間狀態，它代表需要檢查消息隊列來確定狀態。
###4.6.1 發送事務消息
1) 創建事務性生產者
使用?TransactionMQProducer類創建生產者，并指定唯一的?ProducerGroup，就可以設置自定義線程池來處理這些檢查請求。執行本地事務后、需要根據執行結果對消息隊列進行回復。回傳的事務狀態在請參考前一節。
public class Producer {public static void main(String[] args) throws MQClientException, InterruptedException {//創建事務監聽器TransactionListener transactionListener = new TransactionListenerImpl();//創建消息生產者TransactionMQProducer producer = new TransactionMQProducer("group6");producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");//生產者這是監聽器producer.setTransactionListener(transactionListener);//啟動消息生產者producer.start();String[] tags = new String[]{"TagA", "TagB", "TagC"};for (int i = 0; i < 3; i++) {try {Message msg = new Message("TransactionTopic", tags[i % tags.length], "KEY" + i,("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));SendResult sendResult = producer.sendMessageInTransaction(msg, null);System.out.printf("%s%n", sendResult);TimeUnit.SECONDS.sleep(1);} catch (MQClientException | UnsupportedEncodingException e) {e.printStackTrace();}}//producer.shutdown();}
}
2）實現事務的監聽接口
當發送半消息成功時，我們使用?executeLocalTransaction?方法來執行本地事務。它返回前一節中提到的三個事務狀態之一。checkLocalTranscation?方法用于檢查本地事務狀態，并回應消息隊列的檢查請求。它也是返回前一節中提到的三個事務狀態之一。
public class TransactionListenerImpl implements TransactionListener {@Overridepublic LocalTransactionState executeLocalTransaction(Message msg, Object arg) {System.out.println("執行本地事務");if (StringUtils.equals("TagA", msg.getTags())) {return LocalTransactionState.COMMIT_MESSAGE;} else if (StringUtils.equals("TagB", msg.getTags())) {return LocalTransactionState.ROLLBACK_MESSAGE;} else {return LocalTransactionState.UNKNOW;}}@Overridepublic LocalTransactionState checkLocalTransaction(MessageExt msg) {System.out.println("MQ檢查消息Tag【"+msg.getTags()+"】的本地事務執行結果");return LocalTransactionState.COMMIT_MESSAGE;}
}
4.6.2 使用限制
- 事務消息不支持延時消息和批量消息。
- 為了避免單個消息被檢查太多次而導致半隊列消息累積，我們默認將單個消息的檢查次數限制為 15 次，但是用戶可以通過 Broker 配置文件的?
transactionCheckMax參數來修改此限制。如果已經檢查某條消息超過 N 次的話（ N =?transactionCheckMax?） 則 Broker 將丟棄此消息，并在默認情況下同時打印錯誤日志。用戶可以通過重寫?AbstractTransactionCheckListener?類來修改這個行為。 - 事務消息將在 Broker 配置文件中的參數 transactionMsgTimeout 這樣的特定時間長度之后被檢查。當發送事務消息時，用戶還可以通過設置用戶屬性 CHECK_IMMUNITY_TIME_IN_SECONDS 來改變這個限制，該參數優先于?
transactionMsgTimeout?參數。 - 事務性消息可能不止一次被檢查或消費。
- 提交給用戶的目標主題消息可能會失敗，目前這依日志的記錄而定。它的高可用性通過 RocketMQ 本身的高可用性機制來保證，如果希望確保事務消息不丟失、并且事務完整性得到保證，建議使用同步的雙重寫入機制。
- 事務消息的生產者 ID 不能與其他類型消息的生產者 ID 共享。與其他類型的消息不同，事務消息允許反向查詢、MQ服務器能通過它們的生產者 ID 查詢到消費者。
)
![[翻譯] Visual Studio 2019 RC版發布](http://pic.xiahunao.cn/[翻譯] Visual Studio 2019 RC版發布)
