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.9.4

下载地址

2.2.2 环境要求

Linux64位系统
JDK1.8(64位)
源码安装需要安装Maven 3.5.4

2.2 安装RocketMQ

2.2.1 安装步骤

本教程以二进制包方式安装

1.上传压缩包到linux

解压安装 jdk

shell

tar -zxvf jdk-8u221-linux-x64.tar.gz -C /usr/local
cd /usr/local
mv jdk1.8.0_221 jdk
vim /etc/profile
# 在环境变量文件中，添加如下内容
export JAVA_HOME=/usr/local/jdk
export PATH=$JAVA_HOME/bin:$PATH
# 重新加载环境变量文件
source /etc/profile
# 最终测试
java -version

3.解压rocketmq 压缩包

shell

unzip rocketmq-all-4.9.4-bin-release.zip
mv rocketmq-all-4.9.4-bin-release /usr/local/rocketmq     #把它剪切到   /usr/local/rocketmq 下面

2.2.2 目录介绍

benchmark 里面放了一些demo 快速演示案例
bin：启动脚本，包括shell脚本和CMD脚本
conf：实例配置文件，包括broker配置文件、logback配置文件等
lib：依赖jar包，包括Netty、commons-lang、FastJSON等

2.3 启动RocketMQ 改好内存设置以后再启动

启动NameServer

shell

# 1.进入 bin目录    启动NameServem        nohup命令： 不挂起  用于在系统后台不挂断地运行命令，退出终端不会影响程序的运行   
nohup sh ./mqnamesrv &
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/namesrv.log

启动Broker

shell

# 1.启动Broker
nohup sh ./mqbroker -n localhost:9876 &
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/broker.log

问题描述：
RocketMQ默认的虚拟机内存较大，启动Broker如果因为内存不足失败，需要编辑如下两个配置文件，修改JVM内存大小

shell

# 编辑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.设置临时环境变量    在  bin 目录 执行下面的命令    
export NAMESRV_ADDR=localhost:9876
# 2.使用安装包的Demo发送消息
sh ./tools.sh org.apache.rocketmq.example.quickstart.Producer   # 生产者发送消息  发完消息就停下

2.4.2 接收消息

shell

# 1.设置临时环境变量    在  bin 目录 执行下面的命令
export NAMESRV_ADDR=localhost:9876
# 2.接收消息
sh ./tools.sh org.apache.rocketmq.example.quickstart.Consumer    # 消费者消费消息     消费者会一直处于监听状态 等着接受消息

2.5 关闭RocketMQ

shell

# 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 （High Availability 高可用）采用异步复制方式，主备有短暂消息延迟（毫秒级），这种模式的优缺点如下：

优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时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 服务器环境

按照上面搭建单节点rocketmq 的过程，准备两台服务器即都安装好 jdk 和 rocketmq,然后进行下面的集群配置

序号	IP	角色	架构模式
1	192.168.25.135	nameserver、brokerserver	Master1、Slave2
2	192.168.25.138	nameserver、brokerserver	Master2、Slave1

3.3.4 Host添加信息

bash

vim /etc/hosts

配置如下:

bash


# nameserver
192.168.211.101 rocketmq-nameserver1
192.168.211.102 rocketmq-nameserver2
# broker
192.168.211.101 rocketmq-master1
192.168.211.101 rocketmq-slave2
192.168.211.102 rocketmq-master2
192.168.211.102 rocketmq-slave1

配置完成后, 重启网卡

bash

systemctl restart network

3.3.5 防火墙配置

宿主机需要远程访问虚拟机的rocketmq服务和web服务，需要开放相关的端口号，简单粗暴的方式是直接关闭防火墙

bash

# 关闭防火墙
systemctl stop firewalld.service 
# 查看防火墙的状态
firewall-cmd --state 
# 禁止firewall开机启动
systemctl disable firewalld.service

或者为了安全，只开放特定的端口号，RocketMQ默认使用3个端口：9876 、10911 、11011 。如果防火墙没有关闭的话，那么防火墙就必须开放这些端口：

nameserver 默认使用 9876 端口
master 默认使用 10911 端口
slave 默认使用11011 端口

执行以下命令：

bash

# 开放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 环境变量配置

bash

vim /etc/profile

在profile文件的末尾加入如下命令

bash

#set rocketmq
ROCKETMQ_HOME=/usr/local/rocketmq
PATH=$PATH:$ROCKETMQ_HOME/bin
export ROCKETMQ_HOME PATH

输入:wq! 保存并退出，并使得配置立刻生效：

bash

source /etc/profile

3.3.7 创建消息存储路径

bash

#  192.168.211.101上  存储    master1 
mkdir /usr/local/rocketmq/store1
mkdir /usr/local/rocketmq/store1/commitlog
mkdir /usr/local/rocketmq/store1/consumequeue
mkdir /usr/local/rocketmq/store1/index
# 192.168.211.101 上   存储   slave2
mkdir /usr/local/rocketmq/store2
mkdir /usr/local/rocketmq/store2/commitlog
mkdir /usr/local/rocketmq/store2/consumequeue
mkdir /usr/local/rocketmq/store2/index


#  192.168.211.102上  存储    slave1
mkdir /usr/local/rocketmq/store1
mkdir /usr/local/rocketmq/store1/commitlog
mkdir /usr/local/rocketmq/store1/consumequeue
mkdir /usr/local/rocketmq/store1/index
# 192.168.211.102 上   存储   master2
mkdir /usr/local/rocketmq/store2
mkdir /usr/local/rocketmq/store2/commitlog
mkdir /usr/local/rocketmq/store2/consumequeue
mkdir /usr/local/rocketmq/store2/index

3.3.8 broker配置文件

1）master1

服务器：192.168.211.101

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a.properties

修改配置如下：

bash

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=0
# 如果存在docker环境时  需要设置brokerIP1  要不然 broker 会走容器ip
brokerIP1=192.168.211.101
#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/store1
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store1/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store1/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store1/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store1/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store1/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.211.101

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b-s.properties

修改配置如下：

bash

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0 表示 Master，>0 表示 Slave
brokerId=1
# 如果存在docker环境时  需要设置brokerIP1
brokerIP1=192.168.211.101
#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/store2
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store2/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store2/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store2/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store2/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store2/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.211.102

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b.properties

修改配置如下：

bash

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0 表示 Master，>0 表示 Slave
brokerId=0
# 存在docker环境时  需要设置brokerIP1
brokerIP1=192.168.25.138
#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/store2
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store2/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store2/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store2/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store2/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store2/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.211.105

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a-s.properties

修改配置如下：

bash

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=1
# 存在docker环境时  需要设置brokerIP1
brokerIP1=192.168.25.138
#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/store1
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store1/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store1/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store1/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store1/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store1/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参数进行调整：

bash

#===================================================
# 开发环境配置 JVM Configuration
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"

####2）runserver.sh

vim /usr/local/rocketmq/bin/runserver.sh

bash

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

3.3.10 服务启动

1）启动NameServe集群

分别在192.168.211.101和192.168.211.102启动NameServer

bash

cd /usr/local/rocketmq/bin
nohup sh mqnamesrv &

2）启动Broker集群

在192.168.211.101上启动master1和slave2

master1：

bash

cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-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.211.102上启动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
getBrokerConfig	获取Broker配置	-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
getNamesrvConfig	获取NameServer配置	-h	打印帮助
updateNamesrvConfig	修改NameServer配置	-n	NameServer 服务地址，格式 ip:port
		-h	打印帮助
		-k	key
		-v	value

8）其他

名称	含义	命令选项	说明
startMonitoring	开启监控进程，监控消息误删、重试队列消息数等	-n	NameServer 服务地址，格式 ip:port
startMonitoring	开启监控进程，监控消息误删、重试队列消息数等	-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进行编译打包运行。

**但是新版本不叫 rocketmq-console 了叫 rockermq-dashboard **

3.5.2 下载并编译打包

git clone https://github.com/apache/rocketmq-externals
cd rocketmq-console
mvn clean package -Dmaven.test.skip=true

注意：打包前在rocketmq-console中配置namesrv集群地址：并把生成的jar 放在 rocketmq 服务器上

rocketmq.config.namesrvAddr=192.168.211.101:9876;192.168.211.102:9876

启动rocketmq-console：

java -jar rocketmq-console-ng-1.0.0.jar

我在启动的时候报了一个这样的错误

java

java.lang.RuntimeException: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to <172.20.0.1:10909> failed
# 解决方案  是把 两台虚拟机的  10909 端口都打开

启动成功后，我们就可以通过浏览器访问http://192.168.211.101:8080进入控制台界面了，如下图：

集群状态：

3.6 windows 版本的 rocketmq 安装

3.6.1 配置 rocketmq_home

把ROCKETMQ_HOME配置path 中

3.6.2 配置 nameserver.properties

properties

listenPort=9876
listenIp=127.0.0.1

启动nameserver

shell

start  mqnamesrv.cmd -c  ../conf/nameserver.properties

如果服务器性能不够，修改runserver.cmd里面jvm配置

nameserver进程启动成功

启动broker进程

配置broker

properties

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=127.0.0.1:9876
#Broker 对外服务的监听端口
listenPort=10911
#Broker监听的ip
brokerIP1=127.0.0.1

#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000


enablePropertyFilter=true
#事务消息
transactionTimeout=3000
transactionCheckMax=5
transactionCheckInterval=2000

启动broker

在bin目录下，新建my-broker.cmd文件，启动内容如下:

shell

start mqbroker.cmd -c  ../conf/broker.conf

broker启动成功

注意： mq 的数据是持久化的保存在硬盘上的，这个保存位置咱们可以进行配置，如果不进行配置默认是在 c盘 C:\Users\Administrator\store

有时遇到 broker 启动不了，可以尝试删除这个目录

4. 消息发送样例

导入MQ客户端依赖

xml

  <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.9.4</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>

消息发送者步骤分析

tex

1.创建消息生产者producer，并制定生产者组名
2.指定Nameserver地址
3.启动producer
4.创建消息对象，指定主题Topic、Tag和消息体
5.发送消息
6.关闭生产者producer

消息消费者步骤分析

tex

1.创建消费者Consumer，制定消费者组名
2.指定Nameserver地址
3.订阅主题Topic和Tag
4.设置回调函数，处理消息
5.启动消费者consumer

4.1 基本样例

4.1.1 消息发送

1）发送同步消息

这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。

java

public class SyncProducer {
	public static void main(String[] args) throws Exception {
    	// 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("producer-group1");
    	// 设置NameServer的地址
    	producer.setNamesrvAddr("localhost:9876");
    	// 启动Producer实例
        producer.start();
    	for (int i = 0; i < 10; i++) {
    	    // 创建消息，并指定Topic，Tag和消息体
    	    Message msg = new Message("Demo1Topic" /* Topic */,
        	"Demo1Tag" /* Tag */,
        	("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        	);
        	// 发送消息到一个Broker
            SendResult sendResult = producer.send(msg);
            // 通过sendResult返回消息是否成功送达
            System.out.printf("%s%n", sendResult);
    	}
    	// 如果不再发送消息，关闭Producer实例。
    	producer.shutdown();
    }
}

2）发送异步消息

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。

java

public class AsyncProducer {
	public static void main(String[] args) throws Exception {
    	// 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("producer-group1");
    	// 设置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("Demo1Topic",
                    "Demo1Tag",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // SendCallback接收异步返回结果的回调
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.printf("%-10d OK %s %n", index,
                            sendResult.getMsgId());
                    }
                    @Override
                    public void onException(Throwable e) {
      	              System.out.printf("%-10d Exception %s %n", index, e);
      	              e.printStackTrace();
                    }
            	});
    	}
    
    }
}

3）单向发送消息

这种方式主要用在不特别关心发送结果的场景，例如日志发送。

java

public class OnewayProducer {
	public static void main(String[] args) throws Exception{
    	// 实例化消息生产者Producer
        DefaultMQProducer 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）负载均衡模式

消费者采用负载均衡方式消费消息，多个消费者共同消费队列消息，每个消费者处理的消息不同

java

public static void main(String[] args) throws Exception {
    // 实例化消息生产者,指定组名
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
    // 指定Namesrv地址信息.
    consumer.setNamesrvAddr("localhost:9876");
    // 订阅Topic   参数1 主题 ,  参数2  tag
    consumer.subscribe("Test", "*");
    //负载均衡模式消费
    consumer.setMessageModel(MessageModel.CLUSTERING);
    // 注册回调函数，处理消息
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public 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）广播模式

消费者采用广播的方式消费消息，每个消费者消费的消息都是相同的

java

public static void main(String[] args) throws Exception {
    // 实例化消息生产者,指定组名
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
    // 指定Namesrv地址信息.
    consumer.setNamesrvAddr("localhost:9876");
    // 订阅Topic
    consumer.subscribe("Test", "*");
    //广播模式消费
    consumer.setMessageModel(MessageModel.BROADCASTING);
    // 注册回调函数，处理消息
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public 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 顺序消息生产

java

package com.glls.rocketmq.order.orderdemo2;

import com.glls.rocketmq.MQCommon;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

/**
 * @date 2023/6/15
 * @desc
 */
public class OrderProducer {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        //1.创建消息生产者DefaultMQProducer，并制定生产者group
        DefaultMQProducer producer = new DefaultMQProducer(MQCommon.PRODUCER_GROUP1);

        //2.指定Nameserver地址
        producer.setNamesrvAddr("localhost:9876");
        //3.启动producer
        producer.start();

        String[] tags = new String[]{"TagA", "TagB", "TagC"};

        List<OrderStep> orderSteps = OrderStep.buildOrders();

        for (int i = 0; i < orderSteps.size(); i++) {

            String body = orderSteps.get(i).toString();

            //4.构造消息对象
            Message message = new Message(MQCommon.DEMO1_TOPIC, tags[i % tags.length],body.getBytes());
            //5.发送消息
            SendResult send = producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {

                    Long id = (Long) arg;  //根据订单id选择发送queue
                    long index = id % mqs.size();
                    return mqs.get((int) index);

                }
            },orderSteps.get(i).getOrderId());

            System.out.println(send);
        }
        producer.shutdown();
    }
}

OrderStep

java

package com.glls.rocketmq.order.orderdemo2;

import java.util.ArrayList;
import java.util.List;

/**
 * @date 2023/6/15
 * @desc
 */
public 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;
    }

    @Override
    public String toString() {
        return "OrderStep{" +
                "orderId=" + orderId +
                ", desc='" + desc + '\'' +
                '}';
    }


    /**
     * 生成模拟订单数据
     */
    public static 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 顺序消费消息

java

package com.glls.rocketmq.order.orderdemo2;

import com.glls.rocketmq.MQCommon;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @date 2023/6/15
 * @desc
 */
public class OrderConsumer {

    public static void main(String[] args) throws  Exception{
        //1.创建消费者Consumer(DefaultMQPushConsumer)，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MQCommon.CONSUMER_GROUP1);
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("localhost:9876");

        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        //3.订阅(subscribe)主题Topic
        consumer.subscribe(MQCommon.DEMO1_TOPIC,"TagA || TagB || TagC");
        //4.设置消费模式(MessageModel),默认负载均衡模式
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //5.注册（register）回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerOrderly() {

            Random random = new Random();
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                try {
                    for (MessageExt messageExt : msgs) {
                        //获取消息的topic
                        String topic = messageExt.getTopic();
                        //获取消息的标签
                        String tags = messageExt.getTags();
                        //获取消息内容
                        String content = new String(messageExt.getBody());
                        System.out.println("topic"+topic+"==tags"+tags+"==content"+content);

                        try {
                            //模拟业务逻辑处理中...
                            TimeUnit.SECONDS.sleep(random.nextInt(5));
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }catch (Exception e){
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }


                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        //6.启动消费者consumer
        consumer.start();
    }
}

总结：顺序消息就是在生产者指定MessageQueueSelector ，消费者去指定 MessageListenerOrderly

4.3 延时消息

比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

4.3.1 启动消息消费者

java

package com.glls.rocketmq.delay;

import com.glls.rocketmq.MQCommon;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Date;
import java.util.List;

/**
 * @date 2023/6/15
 * @desc
 */
public class DelayConsumer {

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MQCommon.CONSUMER_GROUP1);

        consumer.setNamesrvAddr("localhost:9876");

        consumer.subscribe(MQCommon.DEMO1_TOPIC, MQCommon.DEMO1_TAG);

        // 注册消息监听者
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
                for (MessageExt message : messages) {
                    // Print approximate delay time period
                    System.out.println("Receive message[msgId=" + message.getMsgId() + "] " + "当前时间：" + new Date().toLocaleString()+"发送时间："+ new String(message.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();

        System.out.println("消费者启动");
    }
}

4.3.2 发送延时消息

java

package com.glls.rocketmq.delay;

import com.glls.rocketmq.MQCommon;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @date 2023/6/15
 * @desc
 */
public class DelayProducer {

    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer(MQCommon.PRODUCER_GROUP1);

        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        for (int i = 0; i < 10; i++) {
            Date date = new Date();

            String s = date.toLocaleString();

            Message message = new Message(MQCommon.DEMO1_TOPIC, MQCommon.DEMO1_TAG, s.getBytes());
            // 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
            message.setDelayTimeLevel(3);
            // 发送消息
            SendResult send = producer.send(message);

            System.out.println("发送结果"+ send.getSendStatus());

            TimeUnit.SECONDS.sleep(1);
        }
        // 关闭生产者
        producer.shutdown();
    }
}

###4.3.3 验证

您将会看到消息的消费比存储时间晚10秒

4.3.4 使用限制

java

// 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，相同的waitStoreMsg OK，而且不能是延时消息。此外，这一批消息的总大小不应超过4MB。

4.4.1 发送批量消息

如果您每次只发送不超过4MB的消息，则很容易使用批处理，样例如下：

java

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时，这时候最好把消息进行分割

java

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是一个简单而有用的设计，其可以来选择您想要的消息。例如：

java

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

消费者将接收包含TAGA或TAGB或TAGC的消息。但是限制是一个消息只能有一个标签，这对于复杂的场景可能不起作用。在这种情况下，可以使用SQL表达式筛选消息。SQL特性可以通过发送消息时的属性来进行计算。在RocketMQ定义的语法下，可以实现一些简单的逻辑。下面是一个例子：

shell

------------
| 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基本语法

注意：需要在broker.conf 中开启配置

shell

enablePropertyFilter=true

RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。

数值比较，比如：>，>=，<，<=，BETWEEN，=；
字符比较，比如：=，<>，IN；
IS NULL 或者 IS NOT NULL；
逻辑符号 AND，OR，NOT；

常量支持类型为：

数值，比如：123，3.1415；
字符，比如：'abc'，必须用单引号包裹起来；
NULL，特殊的常量
布尔值，TRUE 或 FALSE

只有使用push模式的消费者才能用使用SQL92标准的sql语句，接口如下：

java

public void subscribe(finalString topic, final MessageSelector messageSelector)

4.5.2 消息生产者

发送消息时，你能通过putUserProperty来设置消息的属性

java

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筛选消息

java

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() {
   @Override
   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
   }
});
consumer.start();

4.6 事务消息

4.6.1 流程分析

事务消息 producer 发送给MQ的消息是带有事务控制的，发的消息需要生产者一个事务的提交如果消息发送给MQ Server 但是没有提交事务那么这个消息是不能被消费者消费的，这种由于生产者没有提交而导致消费者不可消费的消息我们称之为 half 消息，half 消息发送完了之后 MQ 会给生产者一个反馈消息已经收到了，然后生产者执行本地的事务，当你把本地事务执行完了之后生产者再对half消息执行一个 Commit 或者 Rollback , 当做了Commit 操作，则消息可以被消费者消费了，如果执行了Rollback ,MQ Server 就会把这个消息删除掉，消费者就接受不到这个消息了，如果在做提交和回滚操作时失败了或者超时了这个时候 MQ Server 会对消息做一个回查就是一个Check Back 通过回调方法检查消息的状态通过检查状态再对消息进行处理（提交或回滚）

上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程。

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，就可以设置自定义线程池来处理这些检查请求。执行本地事务后、需要根据执行结果对消息队列进行回复。回传的事务状态在请参考前一节。

java

package com.glls.rocketmq.transaction;

import com.glls.rocketmq.MQCommon;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @date 2023/6/16
 * @desc 事务消息生产者
 */
public class TransactionProducer {

    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionMQProducer producer = new TransactionMQProducer(MQCommon.PRODUCER_TRANSACTION_GROUP1);

        producer.setNamesrvAddr("127.0.0.1:9876");
        //设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 在这个方法 执行本地事务   本地事务的执行入口
             * 发送半事务消息之后  执行的方法
             * */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                //情况1  本地事务执行成功   进行 commit
                //System.out.println("本地事务执行成功,提交操作");
                //return LocalTransactionState.COMMIT_MESSAGE;

                //情况2
                //System.out.println("本地事务执行失败,回滚操作");
                //return LocalTransactionState.ROLLBACK_MESSAGE;

                //情况3  可能 事务还没执行完  不知道 成功还是 失败
                //return LocalTransactionState.UNKNOW;


                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;
                }
            }

            /**
             * 事务状态回查      在一段时间之后 进行消息回查
             * */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {

                System.out.println("事务回查时间："+  new Date().toLocaleString());
                System.out.println("MQ回查消息的Tag为:"+ msg.getTags() +"本地事务执行结果");
                //return LocalTransactionState.COMMIT_MESSAGE;
                return LocalTransactionState.UNKNOW;
            }
        });



        // 启动producer
        producer.start();
        String[] tags = new String[]{"TagA","TagB","TagC"};
        for (int i = 0; i < 3; i++) {
            Message message = new Message(MQCommon.TRANSACTION_TOPIC, tags[i % tags.length], ("Hello Transaction Msg" + i).getBytes());
            /**
             * 发送事务消息
             * 参数1  消息对象
             * 参数2  将事务控制 应用到某一条消息    这里 咱们传入 null  表示  对每条消息 都进行事务控制
             * */
            TransactionSendResult sendResult = producer.sendMessageInTransaction(message, null);

            System.out.println("发送状态："+sendResult.getSendStatus() +"消息id:"+sendResult.getMsgId() + "发送时间：" + new Date().toLocaleString());

            TimeUnit.SECONDS.sleep(1);
        }
        //producer.shutdown();   这可不能关


    }
}

2）事务消息消费者

java

package com.glls.rocketmq.transaction;

import com.glls.rocketmq.MQCommon;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @date 2023/6/16
 * @desc  事务消息消费者
 */
public class TransactionConsumer {

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MQCommon.CONSUMER_TRANSACTION_GROUP1);

        consumer.setNamesrvAddr("127.0.0.1:9876");

        consumer.subscribe(MQCommon.TRANSACTION_TOPIC,"TagA || TagB || TagC");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                for(MessageExt msg: msgs){
                    System.out.println("消息id:"+msg.getMsgId() +"消息内容："+ new String(msg.getBody()));
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();

        System.out.println("消费者 启动");

    }
}

3）实现事务的监听接口单独定义一个监听器类

当发送半消息成功时，我们使用 executeLocalTransaction 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。checkLocalTranscation 方法用于检查本地事务状态，并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。

java

public class TransactionListenerImpl implements TransactionListener {

    @Override
    public 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;
        }

    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("MQ检查消息Tag【"+msg.getTags()+"】的本地事务执行结果");
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

//回查相关配置是在 broker.conf 配置文件中

            //第一次回查的 时间间隔    3秒
            //transactionTimeout=3000
            //回查次数  默认15 次  这里设置成5次   达到最大回查次数 仍然没有返回 提交 或者 回滚状态， 则 按照回滚处理
            //transactionCheckMax=5
            //每隔两秒  回查一次
            //transactionCheckInterval=2000

4.6.2 使用限制

事务消息不支持延时消息和批量消息。
为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的 transactionCheckMax参数来修改此限制。如果已经检查某条消息超过 N 次的话（ N = transactionCheckMax ）则 Broker 将丢弃此消息，并在默认情况下同时打印错误日志。用户可以通过重写 AbstractTransactionCheckListener 类来修改这个行为。
事务消息将在 Broker 配置文件中的参数 transactionMsgTimeout 这样的特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionMsgTimeout 参数。
事务性消息可能不止一次被检查或消费。
提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。
事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。

4.7：重试队列

RocketMQ消费端默认有重试机制

消费端重试分为两种情况

异常重试：由于Consumer端逻辑出现了异常，导致返回了RECONSUME_LATER状态，那么Broker就会在一段时间后尝试重试。
超时重试：如果Consumer端处理时间过长，或者由于某些原因线程挂起，导致迟迟没有返回消费状态，Broker就会认为Consumer消费超时，此时会发起超时重试。

重试队列名称为：%RETRY%+consumergroup

设置重试时间与次数：

可在broker.conf文件中配置Consumer端的重试次数和重试时间间隔

第几次重试	与上次重试的间隔时间	第几次重试	与上次重试的间隔时间
1	10 秒	9	7 分钟
2	30 秒	10	8 分钟
3	1 分钟	11	9 分钟
4	2 分钟	12	10 分钟
5	3 分钟	13	20 分钟
6	4 分钟	14	30 分钟
7	5 分钟	15	1 小时
8	6 分钟	16	2 小时

如果消息重试 16 次后仍然失败，消息将不再投递。如果严格按照上述重试时间间隔计算，某条消息在一直消费失败的前提下，将会在接下来的 4 小时 46 分钟之内进行 16 次重试，超过这个时间范围消息将不再重试投递。

注意：一条消息无论重试多少次，这些重试消息的 Message ID 不会改变。

重试配置

集群消费方式下，消息消费失败后期望消息重试，需要在消息监听器接口的实现中明确进行配置（三种方式任选一种）：

返回 RECONSUME_LATER （推荐）
返回 Null
抛出异常

异常重试演示：

java

package com.glls.rocketmq.retry;

import com.glls.rocketmq.MQCommon;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Date;
import java.util.List;

/**
 * @date 2023/6/16
 * @desc
 */
public class RetryConsumer {

    public static void main(String[] args) throws MQClientException {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MQCommon.CONSUMER_GROUP1);

        consumer.setNamesrvAddr(MQCommon.NAMESERVER);

        consumer.subscribe(MQCommon.DEMO1_TOPIC,MQCommon.DEMO1_TAG);


        //设置消息的最大重试次数  当消息重试达到最大次数 消息会被发送到死信队列   默认在死信队列存三天
        consumer.setMaxReconsumeTimes(2);

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                for (int i = 0; i <msgs.size() ; i++) {


                    try {
                        //获取消息的Topic
                        String topic = msgs.get(i).getTopic();
                        //获取消息的Tag
                        String tags = msgs.get(i).getTags();

                        //获取消息的内容
                        byte[] body = msgs.get(i).getBody();

                        String msg = new String(body);

                        // Print approximate delay time period
                        System.out.println("Receive message[msgId=" + msgs.get(i).getMsgId() + "] " + "当前时间:"+ new Date().toLocaleString() + "发送时间"+ new String(msgs.get(i).getBody()) );
                        int j = 5/0;
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }


                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;


            }
        });
        //5.启动消费者consumer
        consumer.start();

        System.out.println("消费者已启动");


    }
}

java

package com.glls.rocketmq.retry;

import com.glls.rocketmq.MQCommon;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.Date;

/**
 * @date 2023/6/16
 * @desc 消息重试的生产者
 */
public class RetryProducer {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer(MQCommon.PRODUCER_GROUP1);

        producer.setNamesrvAddr(MQCommon.NAMESERVER);

        producer.start();

        Date date = new Date();
        String s = date.toLocaleString();
        Message message = new Message(MQCommon.DEMO1_TOPIC,MQCommon.DEMO1_TAG, s.getBytes());

        // 发送消息
        producer.send(message);
        System.out.println("生产者已发送");
        producer.shutdown();
    }
}

注意1：只有在消息模式为MessageModel.CLUSTERING集群模式时，Broker才会自动进行重试，广播消息是不会重试的。
注意2：由于MQ的重试机制，难免会引起消息的重复消费问题。比如一个ConsumerGroup中有两个，Consumer1和Consumer2，以集群方式消费。假设一条消息发往ConsumerGroup，由Consumer1消费，但是由于Consumer1消费过慢导致超时，如果Broker将消息发送给Consumer2去消费，这样就产生了重复消费问题。因此，使用MQ时应该对一些关键消息进行幂等去重的处理。

delete from t_account where id = 1;

update t_account set money = money + 200 where id =1;

面试题：怎么保证消息消费的时候0丢失？

答：Rocketmq默认就有重试机制，如果第一次消费的时候，broker收到的回应是（ConsumeConcurrentlyStatus.RECONSUME_LATER），那么这条消息不会丢失，会进入重试队列，重试的次数与重试的时间可以配置，如果重试的次数超过配置的次数，那么这条消息没有丢失，但是放入死信队列

面试题：怎么解决重试幂等性问题？

答：如果重试很有可能出现幂等性问题，需要业务逻辑做配合，比如可以先判断数据库的状态，然后根据数据库的状态，做对应的处理

其实就是占坑思想，比如 setnx 多个消费者只有第一个消费者占坑成功可以消费

4.8：死信队列

txt

当一条消息初次消费失败，消息队列 RocketMQ 会自动进行消息重试，达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息。此时，消息队列 RocketMQ 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。

在消息队列 RocketMQ 版中，这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。


死信消息具有以下特性：
1:不会再被之前的消费者正常消费。
2:有效期与正常消息相同，均为 3 天，3 天后会被自动删除。因此，请在死信消息产生后的 3 天内及时处理。


死信队列具有以下特性：
1:一个死信队列对应一个 Group ID， 而不是对应单个消费者实例。
2:如果一个 Group ID 未产生死信消息，消息队列 RocketMQ 便不会为其创建相应的死信队列。
3:一个死信队列包含了对应 Group ID 产生的所有死信消息，不论该消息属于哪个 Topic。
4:消息队列 RocketMQ 版控制台提供对死信消息的查询、重发的功能。

java

package com.glls.rocketmq.retry;

/**
 * @date 2023/6/16
 * @desc
 */

import com.glls.rocketmq.MQCommon;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Date;
import java.util.List;

/**
 * @date 2023-03-06
 * @desc  死信队列的消费者  可以拿到 死信消息
 * 在这里  可以实现 一些  补偿措施
 *
 * 比如  充值点券 等场景  用户的钱扣了  点券没有充上   就可以 在这个 死信队列的消费者
 * 把 相关的业务数据  存到数据库
 */
public class ConsumerDLQ {

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MQCommon.CONSUMER_GROUP1);
        consumer.setNamesrvAddr("127.0.0.1:9876");

        consumer.subscribe("%DLQ%consumer-group1",MQCommon.DEMO1_TAG);


        //设置消息的最大重试次数  当消息重试达到最大次数 消息会被发送到死信队列   默认在死信队列存三天
        consumer.setMaxReconsumeTimes(3);

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                for (int i = 0; i <msgs.size() ; i++) {


                    try {
                        //获取消息的Topic
                        String topic = msgs.get(i).getTopic();
                        //获取消息的Tag
                        String tags = msgs.get(i).getTags();

                        //获取消息的内容
                        byte[] body = msgs.get(i).getBody();

                        String msg = new String(body);

                        // Print approximate delay time period
                        System.out.println("Receive message[msgId=" + msgs.get(i).getMsgId() + "] " + "当前时间:"+ new Date().toLocaleString() + "发送时间"+ new String(msgs.get(i).getBody()) );

                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }


                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;


            }
        });
        //5.启动消费者consumer
        consumer.start();

        System.out.println("消费者已启动");
    }
}

1. MQ介绍 ​

1.2 MQ的优点和缺点 ​

1.3 各种MQ产品的比较 ​

2. RocketMQ快速入门 ​

2.1 准备工作 ​

2.1.1 下载RocketMQ ​

2.2.2 环境要求 ​

2.2 安装RocketMQ ​

2.2.1 安装步骤 ​

2.2.2 目录介绍 ​

2.3 启动RocketMQ 改好内存设置以后再启动 ​

2.4 测试RocketMQ ​

2.4.1 发送消息 ​

2.4.2 接收消息 ​

2.5 关闭RocketMQ ​

3. RocketMQ集群搭建 ​

3.1 各角色介绍 ​

3.2 集群搭建方式 ​

3.2.1 集群特点 ​

3.2.3 集群模式 ​

1）单Master模式 ​

2）多Master模式 ​

3）多Master多Slave模式（异步） ​

4）多Master多Slave模式（同步） ​

3.3 双主双从集群搭建 ​

3.3.1 总体架构 ​

3.3.2 集群工作流程 ​

3.3.3 服务器环境 ​

3.3.4 Host添加信息 ​

3.3.5 防火墙配置 ​

3.3.6 环境变量配置 ​

3.3.7 创建消息存储路径 ​

3.3.8 broker配置文件 ​

1）master1 ​

2）slave2 ​

3）master2 ​

4）slave1 ​

3.3.9 修改启动脚本文件 ​

1）runbroker.sh ​

3.3.10 服务启动 ​

1）启动NameServe集群 ​

2）启动Broker集群 ​

3.3.11 查看进程状态 ​

3.3.12 查看日志 ​

3.4 mqadmin管理工具 ​

3.4.1 使用方式 ​

3.4.2 命令介绍 ​

1）Topic相关 ​

2）集群相关 ​

3）Broker相关 ​

4）消息相关 ​

5）消费者、消费组相关 ​

6）连接相关 ​

7）NameServer相关 ​

8）其他 ​

3.4.3 注意事项 ​

3.5 集群监控平台搭建 ​

3.5.1 概述 ​

3.5.2 下载并编译打包 ​

3.6 windows 版本的 rocketmq 安装 ​

4. 消息发送样例 ​

4.1 基本样例 ​

4.1.1 消息发送 ​

1）发送同步消息 ​

2）发送异步消息 ​

3）单向发送消息 ​

4.1.2 消费消息 ​

1）负载均衡模式 ​

2）广播模式 ​

4.2 顺序消息 ​

4.2.1 顺序消息生产 ​

4.2.2 顺序消费消息 ​

4.3 延时消息 ​

4.3.1 启动消息消费者 ​

4.3.2 发送延时消息 ​

4.3.4 使用限制 ​

4.4 批量消息 ​

4.4.1 发送批量消息 ​

4.5 过滤消息 ​

4.5.1 SQL基本语法 ​

1. MQ介绍

1.2 MQ的优点和缺点

1.3 各种MQ产品的比较

2. RocketMQ快速入门

2.1 准备工作

2.1.1 下载RocketMQ

2.2.2 环境要求

2.2 安装RocketMQ

2.2.1 安装步骤

2.2.2 目录介绍

2.3 启动RocketMQ 改好内存设置以后再启动

2.4 测试RocketMQ

2.4.1 发送消息

2.4.2 接收消息

2.5 关闭RocketMQ

3. RocketMQ集群搭建

3.1 各角色介绍

3.2 集群搭建方式

3.2.1 集群特点

3.2.3 集群模式

1）单Master模式

2）多Master模式

3）多Master多Slave模式（异步）

4）多Master多Slave模式（同步）

3.3 双主双从集群搭建

3.3.1 总体架构

3.3.2 集群工作流程

3.3.3 服务器环境

3.3.4 Host添加信息

3.3.5 防火墙配置

3.3.6 环境变量配置

3.3.7 创建消息存储路径

3.3.8 broker配置文件

1）master1

2）slave2

3）master2

4）slave1

3.3.9 修改启动脚本文件

1）runbroker.sh

3.3.10 服务启动

1）启动NameServe集群

2）启动Broker集群

3.3.11 查看进程状态

3.3.12 查看日志

3.4 mqadmin管理工具

3.4.1 使用方式

3.4.2 命令介绍

1）Topic相关

2）集群相关

3）Broker相关

4）消息相关

5）消费者、消费组相关

6）连接相关

7）NameServer相关

8）其他

3.4.3 注意事项

3.5 集群监控平台搭建

3.5.1 概述

3.5.2 下载并编译打包

3.6 windows 版本的 rocketmq 安装

4. 消息发送样例

4.1 基本样例

4.1.1 消息发送

1）发送同步消息

2）发送异步消息

3）单向发送消息

4.1.2 消费消息

1）负载均衡模式

2）广播模式

4.2 顺序消息

4.2.1 顺序消息生产

4.2.2 顺序消费消息

4.3 延时消息

4.3.1 启动消息消费者

4.3.2 发送延时消息

4.3.4 使用限制

4.4 批量消息

4.4.1 发送批量消息

4.5 过滤消息

4.5.1 SQL基本语法