RocketMQ学习（一）

文章目录

参考
1. MQ 介绍
- 1.1 为什么要用 MQ
- - 应用解耦
  - 流量削峰
  - 数据分发
- 1.2 MQ 的优点和缺点
- 1.3 各种 MQ 产品的比较
2. RocketMQ 快速入门
- 2.1 准备工作
- - 2.1.1 下载 RocketMQ
  - 2.2.2 环境要求
  - - 安装jdk
- 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
    - 2）runserver.sh
  - 3.3.10 服务启动
  - - 1）启动NameServe集群
    - 2）启动Broker集群
    - - 报错：RocketMQ集群启动报错
      - 示例
  - 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 下载并编译打包
  - 示例
4. 消息发送样例
- 基本步骤
- - 导入 MQ 客户端依赖
  - 消息发送者步骤
  - 消息消费者步骤
- 4.1 基本样例
- - 4.1.1 消息发送
  - - 1）发送同步消息
    - 2）发送异步消息
    - 3）单向发送消息
  - 4.1.2 消费消息
  - - 1）负载均衡模式
    - 2）广播模式
    - 示例
    - - 示例1
      - 示例2
- 4.2 顺序消息
- - 4.2.1 顺序消息生产
  - 4.2.2 顺序消费消息
  - 示例
  - - Producer
    - Consumer
    - 日志输出
- 4.3 延时消息
- - 4.3.1 启动消息消费者
  - 4.3.2 发送延时消息
  - 4.3.4 使用限制
- 4.4 批量消息
- - 4.4.1 发送批量消息
  - - 示例 demo
    - - Consumer
      - Producer
      - 测试
- 4.5 过滤消息
- - 4.5.1 SQL 基本语法
  - 4.5.2 消息生产者
  - 4.5.3 消息消费者
  - - Tag 过滤示例 Demo
    - - Producer
      - Consumer
      - 测试
    - SQL 语法过滤示例 Demo
    - - 问题
      - Producer
      - Consumer
      - 测试
- 4.6 事务消息
- - 4.6.1 流程分析
  - - 1）事务消息发送及提交
    - 2）事务补偿
    - 3）事务消息状态
    - 1) 创建事务性生产者
    - 2）实现事务的监听接口
  - 4.6.2 使用限制
  - 事务消息示例 Demo
  - - Producer
    - Consumer
    - 测试

参考

RocketMq中文官网

RocketMQ的下载与安装（全网最细保姆级别教学）

【SpringBoot高级篇】SpringBoot集成RocketMQ消息队列

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

https://www.apache.org/ 下载地址

在这里插入图片描述

rocketmq中文官网

在这里插入图片描述

2.2.2 环境要求

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

安装jdk

参考：Linux安装JDK完整步骤

1、检查一下系统中的jdk版本

[root@localhost software]# java -version

显示：

openjdk version "1.8.0_102"
OpenJDK Runtime Environment (build 1.8.0_102-b14)
OpenJDK 64-Bit Server VM (build 25.102-b14, mixed mode)

2、检测jdk安装包

[root@localhost software]# rpm -qa | grep java

显示：

java-1.7.0-openjdk-1.7.0.111-2.6.7.8.el7.x86_64
python-javapackages-3.4.1-11.el7.noarch
tzdata-java-2016g-2.el7.noarch
javapackages-tools-3.4.1-11.el7.noarch
java-1.8.0-openjdk-1.8.0.102-4.b14.el7.x86_64
java-1.8.0-openjdk-headless-1.8.0.102-4.b14.el7.x86_64
java-1.7.0-openjdk-headless-1.7.0.111-2.6.7.8.el7.x86_64

3、卸载openjdk

[root@localhost software]# rpm -e --nodeps tzdata-java-2016g-2.el7.noarch
[root@localhost software]# rpm -e --nodeps java-1.7.0-openjdk-1.7.0.111-2.6.7.8.el7.x86_64
[root@localhost software]# rpm -e --nodeps java-1.7.0-openjdk-headless-1.7.0.111-2.6.7.8.el7.x86_64
[root@localhost software]# rpm -e --nodeps java-1.8.0-openjdk-1.8.0.102-4.b14.el7.x86_64
[root@localhost software]# rpm -e --nodeps java-1.8.0-openjdk-headless-1.8.0.102-4.b14.el7.x86_64

或者使用

[root@localhost jvm]# yum remove *openjdk*

之后再次输入rpm -qa | grep java 查看卸载情况：

[root@localhost software]# rpm -qa | grep java
python-javapackages-3.4.1-11.el7.noarch
javapackages-tools-3.4.1-11.el7.noarch

4、安装新的jdk

首先到jdk官网上下载你想要的jdk版本，下载完成之后将需要安装的jdk安装包放到Linux系统指定的文件夹下，并且命令进入该文件夹下：

[root@localhost software]# ll
total 252664
-rw-r--r--. 1 root root  11830603 Jun  9 06:43 alibaba-rocketmq-3.2.6.tar.gz
-rw-r--r--. 1 root root  43399561 Jun  9 06:42 apache-activemq-5.11.1-bin.tar.gz
-rwxrw-rw-. 1 root root 185540433 Apr 21 09:06 jdk-8u131-linux-x64.tar.gz
-rw-r--r--. 1 root root   1547695 Jun  9 06:44 redis-3.2.9.tar.gz
-rw-r--r--. 1 root root  16402010 Jun  9 06:40 zookeeper-3.4.5.tar.gz

解压 jdk-8u131-linux-x64.tar.gz安装包

[root@localhost software]# mkdir -p /usr/lib/jvm
[root@localhost software]# tar -zxvf jdk-8u131-linux-x64.tar.gz -C /usr/lib/jvm

5、设置环境变量

[root@localhost software]# vim /etc/profile

在最前面添加：

export JAVA_HOME=/usr/lib/jvm/jdk1.8.0_221/usr/lib/jvm/jdk1.8.0_131  
export JRE_HOME=${JAVA_HOME}/jre  
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib  
export  PATH=${JAVA_HOME}/bin:$PATH

6、执行profile文件

[root@localhost software]# source /etc/profile

这样可以使配置不用重启即可立即生效。

7、检查新安装的jdk

[root@localhost software]# java -version

显示：

java version "1.8.0_131"
Java(TM) SE Runtime Environment (build 1.8.0_131-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.131-b11, mixed mode)

到此为止，整个安装过程结束。

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.查看启动日志（默认的日志文件就是在/home下）
tail -f ~/logs/rocketmqlogs/namesrv.log

启动 Broker

# 1.启动Broker（启动broker时，需要连上nameServer）
nohup sh bin/mqbroker -n localhost:9876 &

# 2.查看启动日志（默认的日志文件就是在/home下）
tail -f ~/logs/rocketmqlogs/broker.log

问题描述：

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

# 编辑runbroker.sh和runserver.sh修改默认JVM大小（比如对应的8g可以改为 256m）
vi runbroker.sh
vi runserver.sh

参考设置：

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

查看是否启动成功

可以通过上述查看的的日志来查看是否启动成功，也可以通过命令jps查看是否启动成功（如果都启动成功了，会出现NamesrvStartup和BrokerStartup）。

示例

上传rocketmq的压缩包到linux中，并解压

在这里插入图片描述

默认conf目录下的配置文件如下

在这里插入图片描述

默认bin目录下的配置文件如下

在这里插入图片描述

修改runbroker.sh和runserver.sh中的jvm内存大小设置

在这里插入图片描述

分别启动nameserver和broker，并查看日志，~表示的是当前用户家目录，由于我是root用户，因此是/root

在这里插入图片描述

查看rocketmq的日志

在这里插入图片描述

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（可以在配置环境变量之后，直接使用mqshutdown命令，配置方式可参考3.3.6）sh bin/mqshutdown namesrv
# 2.关闭 Brokersh bin/mqshutdown broker
# 可以通过jps来查看这2个服务是否关闭了
jps

3. RocketMQ集群搭建

3.1 各角色介绍

Producer：消息的发送者；举例：发信者（询问NameServer获得1个broker的地址去发送消息）
Consumer：消息接收者；举例：收信者（询问NameServer获得1个broker的地址去消费消息）
Broker：暂存和传输消息；举例：邮局（Broker须主动上报自己的信息到NameServer）
NameServer：管理Broker；举例：各个邮局的管理机构
Topic：区分消息的种类；一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者可以订阅一个或者多个Topic消息
Message Queue：相当于是Topic的分区；用于并行发送和接收消息

在这里插入图片描述

3.2 集群搭建方式

3.2.1 集群特点

在这里插入图片描述

NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步
- （在所有的NameServer启动之后，broker会给每1个NameServer上报自己的信息，因此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 环境变量配置

目的是为了在任意目录下都能直接执行rocketmq下的bin目录中的命令

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配置文件

配置文件在rocketmq解压的文件夹的conf目录下的2m-2s-sync文件夹中，这里配置的步骤就是在192.168.25.135服务器上配置master1和slave2，在192.168.25.138服务器上配置master2和slave1
注意这里面一共有4个broker（双主双从，且使用同步复制的方式），这些broker连的都是同样的NameServer配置，为防止端口冲突在同一服务器的不同broker监听不同的端口，属于同一组的broker的brokerName是一样的。

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-sync/broker-a.properties &

# 本地虚拟机搭建命令
#nohup sh mqbroker -c /usr/local/rocketmq/rocketmq-all-4.4.0-bin-release/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 &

# 本地虚拟机搭建命令
#nohup sh mqbroker -c /usr/local/rocketmq/rocketmq-all-4.4.0-bin-release/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 &

# 本地虚拟机搭建命令
#nohup sh mqbroker -c /usr/local/rocketmq/rocketmq-all-4.4.0-bin-release/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 &

# 本地虚拟机搭建命令
#nohup sh mqbroker -c /usr/local/rocketmq/rocketmq-all-4.4.0-bin-release/conf/2m-2s-sync/broker-a-s.properties &

报错：RocketMQ集群启动报错

参考：https://blog.csdn.net/TaylorSwiftiiln/article/details/121077705，slave启动需要单独的目录，因此创建一个与store同样的目录结构store-s，然后把slave的配置文件的路径改成对应的store-s即可，然后启动

其中，内容如下：

启动RocketMQ集群后报错：

java.lang.RuntimeException: Lock failed,MQ already started
	at org.apache.rocketmq.store.DefaultMessageStore.start(DefaultMessageStore.java:214)
	at org.apache.rocketmq.broker.BrokerController.start(BrokerController.java:827)
	at org.apache.rocketmq.broker.BrokerStartup.start(BrokerStartup.java:64)
	at org.apache.rocketmq.broker.BrokerStartup.main(BrokerStartup.java:58)

原因：

学习阶段，我们将Master和Slave共同部署在一台机器上，同时配置的storePath又相同，没有进行区分

解决：

为从节点创建存储文件

mkdir /usr/local/rocketmq/store-s
mkdir /usr/local/rocketmq/store-s/commitlog
mkdir /usr/local/rocketmq/store-s/consumequeue
mkdir /usr/local/rocketmq/store-s/index

store-s与Master节点用的目录store区分开。

对Slave配置文件进行修改：

#===================storePath主从节点必须分开====================
#存储路径
storePathRootDir=/usr/local/rocketmq/store-s
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store-s/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store-s/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store-s/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store-s/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store-s/abort
#===================storePath主从节点必须分开====================

以上问题解决。

示例

搭建成功

在这里插入图片描述

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 进行编译打包运行。

在这里插入图片描述

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 进入控制台界面了，如下图：

在这里插入图片描述

集群状态：

在这里插入图片描述

示例

可以直接下载这个项目：https://gitee.com/mirrors_apache/rocketmq-dashboard，修改下rocketmq.config.namesrvAddrs=192.168.134.3:9876,192.168.134.4:9876，启动端口改为8899，maven中不用的插件直接注释掉，然后启动项目。即可访问：

在这里插入图片描述

4. 消息发送样例

基本步骤

导入 MQ 客户端依赖

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

消息发送者步骤

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

消息消费者步骤

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

4.1 基本样例

4.1.1 消息发送

1）发送同步消息

这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。（发送完需要阻塞等待 broker 响应）

/**
 * 发送同步消息
 */
public class SyncProducer {

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

        //1.创建消息生产者producer，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");

        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.启动producer
        producer.start();

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

            //4.创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("springboot-mq",              // 消息主题Topic
                                      "Tag1",                       // 消息Tag
                                      ("Hello World" + i).getBytes()// 消息内容
                                     );

            //5.发送消息
            SendResult result = producer.send(msg);

            // 发送状态
            SendStatus status = result.getSendStatus();
            System.out.println("发送结果:" + result);
            
            // 消息id
            String msgId = result.getMsgId();
            // 消息接收队列id
            int queueId = result.getMessageQueue().getQueueId();

            //线程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }

        //6.关闭生产者producer
        producer.shutdown();
    }
}

发送结果

发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A886053CDC18B4AAC2715981050000, offsetMsgId=C0A8860400002A9F0000000000004B5F, messageQueue=MessageQueue [topic=springboot-mq, brokerName=broker-b, queueId=1], queueOffset=0]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A886053CDC18B4AAC27159853E0001, offsetMsgId=C0A8860400002A9F0000000000004C11, messageQueue=MessageQueue [topic=springboot-mq, brokerName=broker-b, queueId=2], queueOffset=0]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A886053CDC18B4AAC27159892D0002, offsetMsgId=C0A8860400002A9F0000000000004CC3, messageQueue=MessageQueue [topic=springboot-mq, brokerName=broker-b, queueId=3], queueOffset=0]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A886053CDC18B4AAC271598D320003, offsetMsgId=C0A8860300002A9F0000000000007C5F, messageQueue=MessageQueue [topic=springboot-mq, brokerName=broker-a, queueId=0], queueOffset=0]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A886053CDC18B4AAC2715991630004, offsetMsgId=C0A8860300002A9F0000000000007D11, messageQueue=MessageQueue [topic=springboot-mq, brokerName=broker-a, queueId=1], queueOffset=0]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A886053CDC18B4AAC2715995500005, offsetMsgId=C0A8860300002A9F0000000000007DC3, messageQueue=MessageQueue [topic=springboot-mq, brokerName=broker-a, queueId=2], queueOffset=0]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A886053CDC18B4AAC27159993F0006, offsetMsgId=C0A8860300002A9F0000000000007E75, messageQueue=MessageQueue [topic=springboot-mq, brokerName=broker-a, queueId=3], queueOffset=0]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A886053CDC18B4AAC271599D420007, offsetMsgId=C0A8860400002A9F0000000000004D75, messageQueue=MessageQueue [topic=springboot-mq, brokerName=broker-b, queueId=0], queueOffset=0]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A886053CDC18B4AAC27159A1350008, offsetMsgId=C0A8860400002A9F0000000000004E27, messageQueue=MessageQueue [topic=springboot-mq, brokerName=broker-b, queueId=1], queueOffset=1]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A886053CDC18B4AAC27159A5250009, offsetMsgId=C0A8860400002A9F0000000000004ED9, messageQueue=MessageQueue [topic=springboot-mq, brokerName=broker-b, queueId=2], queueOffset=1]

2）发送异步消息

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待 Broker 的响应。（就是发送完不需要阻塞等待 broker 的响应）

/**
 * 发送异步消息
 */
public class AsyncProducer {

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

        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");

        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.启动producer
        producer.start();

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

            //4.创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("base", 
                                      "Tag2", 
                                      ("Hello World" + i).getBytes()
                                     );

            //5.发送异步消息（在回调函数中接收发送结果）
            producer.send(msg, new SendCallback() {
                /**
                 * 发送成功回调函数
                 * @param sendResult
                 */
                public void onSuccess(SendResult sendResult) {
                    System.out.println("发送结果：" + sendResult);
                }

                /**
                 * 发送失败回调函数
                 * @param e
                 */
                public void onException(Throwable e) {
                    System.out.println("发送异常：" + e);
                }
            });

            //线程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }

        //6.关闭生产者producer
        producer.shutdown();
    }
}

发送结果

发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A88605353C18B4AAC2715EF09F0000, offsetMsgId=C0A8860300002A9F0000000000007F27, messageQueue=MessageQueue [topic=base, brokerName=broker-a, queueId=1], queueOffset=18]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A88605353C18B4AAC2715EF20F0001, offsetMsgId=C0A8860300002A9F0000000000007FD0, messageQueue=MessageQueue [topic=base, brokerName=broker-a, queueId=3], queueOffset=20]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A88605353C18B4AAC2715EF5F80002, offsetMsgId=C0A8860300002A9F0000000000008079, messageQueue=MessageQueue [topic=base, brokerName=broker-a, queueId=3], queueOffset=21]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A88605353C18B4AAC2715EF9E00003, offsetMsgId=C0A8860300002A9F0000000000008122, messageQueue=MessageQueue [topic=base, brokerName=broker-a, queueId=3], queueOffset=22]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A88605353C18B4AAC2715EFDC80004, offsetMsgId=C0A8860300002A9F00000000000081CB, messageQueue=MessageQueue [topic=base, brokerName=broker-a, queueId=3], queueOffset=23]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A88605353C18B4AAC2715F01B00005, offsetMsgId=C0A8860300002A9F0000000000008274, messageQueue=MessageQueue [topic=base, brokerName=broker-a, queueId=3], queueOffset=24]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A88605353C18B4AAC2715F05990006, offsetMsgId=C0A8860300002A9F000000000000831D, messageQueue=MessageQueue [topic=base, brokerName=broker-a, queueId=3], queueOffset=25]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A88605353C18B4AAC2715F09810007, offsetMsgId=C0A8860300002A9F00000000000083C6, messageQueue=MessageQueue [topic=base, brokerName=broker-a, queueId=0], queueOffset=19]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A88605353C18B4AAC2715F0D690008, offsetMsgId=C0A8860300002A9F000000000000846F, messageQueue=MessageQueue [topic=base, brokerName=broker-a, queueId=2], queueOffset=24]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A88605353C18B4AAC2715F11520009, offsetMsgId=C0A8860300002A9F0000000000008518, messageQueue=MessageQueue [topic=base, brokerName=broker-a, queueId=0], queueOffset=20]

3）单向发送消息

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

/**
 * 发送单向消息
 */
public class OneWayProducer {

    public static void main(String[] args) throws Exception, MQBrokerException {
        
        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        
        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");
        
        //3.启动producer
        producer.start();

        for (int i = 0; i < 3; i++) {
            
            //4.创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("base", 
                                      "Tag3", 
                                      ("Hello World，单向消息" + i).getBytes()
                                     );
            
            //5.发送单向消息
            producer.sendOneway(msg);

            //线程睡1秒
            TimeUnit.SECONDS.sleep(5);
        }

        //6.关闭生产者producer
        producer.shutdown();
    }
}

4.1.2 消费消息

在这里插入图片描述

1）负载均衡模式

消费者采用负载均衡方式消费消息，多个消费者共同消费队列消息，每个消费者处理的消息不同（每一个消息都只会给一个消费者处理）

/**
 * 消息的接受者
 */
public class Consumer {

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

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.订阅主题Topic和Tag
        consumer.subscribe("base", "*");

        //设定消费模式：负载均衡(默认,即不用设置)|广播模式
        consumer.setMessageModel(MessageModel.CLUSTERING);

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 
                                                            ConsumeConcurrentlyContext context) {
                
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

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

2）广播模式

消费者采用广播的方式消费消息，每个消费者消费的消息都是相同的（每一个消息都会发送给所有的消费者）

/**
 * 消息的接受者
 */
public class Consumer {

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

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");

        //3.订阅主题Topic和Tag
        consumer.subscribe("base", "*");

        //设定消费模式：负载均衡(默认,即不用设置)|广播模式
        consumer.setMessageModel(MessageModel.BROADCASTING);

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 
                                                            ConsumeConcurrentlyContext context) {
                
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

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

示例

示例1

消息的生产者发送10次消息到base，Tag1，然后再启动3个Consumer，结果是1个consumer收到（2，6），1个consumer收到（1，5，9），1个消费者收到（0，3，4，7，8），一共收到10条消息

生产者

/**
 * 消息的接受者
 */
public class Consumer {

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

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.订阅主题Topic和Tag
        consumer.subscribe("base", "Tag1");

        //设定消费模式：负载均衡(默认, 此处可以不设置)
        consumer.setMessageModel(MessageModel.CLUSTERING);

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

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

消费者

/**
 * 消息的接受者
 */
public class Consumer {

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

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.订阅主题Topic和Tag
        consumer.subscribe("base", "Tag1");

        //设定消费模式：负载均衡(默认)|广播模式
        consumer.setMessageModel(MessageModel.BROADCASTING);

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

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

示例2

消息的生产者发送10次消息到base，Tag1，然后再启动3个Consumer，结果是3个Consumer都收到了10条消息

生产者

/**
 * 发送同步消息
 */
public class SyncProducer {

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

        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");

        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.启动producer
        producer.start();

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

            //4.创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("base", "Tag1", ("Hello World" + i).getBytes());

            //5.发送消息
            SendResult result = producer.send(msg);

            //发送状态
            SendStatus status = result.getSendStatus();

            System.out.println("发送结果:" + result);

            String msgId = result.getMsgId();
            int queueId = result.getMessageQueue().getQueueId();

            //线程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }

        //6.关闭生产者producer
        producer.shutdown();
    }
}

消费者

/**
 * 消息的接受者
 */
public class Consumer {

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

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.订阅主题Topic和Tag
        consumer.subscribe("base", "Tag1");

        //设定消费模式：负载均衡(默认)|广播模式
        consumer.setMessageModel(MessageModel.BROADCASTING);

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

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

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() {
               @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);
               }
           }, orderList.get(i).getOrderId());//订单id

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

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

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

示例

Producer

@Slf4j
public class Producer {

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

        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");

        //2.指定NameServer地址
        producer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.启动producer
        producer.start();

        //构建消息集合
        List<OrderStep> orderSteps = OrderStep.buildOrders();

        //发送消息（这里一共发送了10条消息）
        for (int i = 0; i < orderSteps.size(); i++) {

            String body = orderSteps.get(i) + "";

            //                               topic        tags    keys      body
            Message message = new Message("OrderTopic", "Order", "i" + i, body.getBytes());

            /**
             * 参数一：消息对象
             * 参数二：消息队列的选择器
             * 参数三：选择队列的业务标识（订单ID）
             */
            SendResult sendResult = producer.send(

                    message,

                    new MessageQueueSelector() {

                        /**
                         *
                         * @param mqs：队列集合
                         * @param msg：消息对象
                         * @param arg：业务标识的参数
                         * @return
                         */
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {

                            long orderId = (long) arg;

                            long index = orderId % mqs.size();

                            return mqs.get((int) index);
                        }
                    },

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

            log.info("发送结果：{}", sendResult);
        }

        producer.shutdown();
    }

}

Consumer

@Slf4j
public class Consumer {

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

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定NameServer地址
        consumer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.订阅主题Topic和Tag
        consumer.subscribe("OrderTopic", "*");

        //4.注册消息监听器（这里使用的是MessageListenerOrderly, 而不是MessageListenerConcurrently。意思就是针对同1个队列的消息的处理使用同1个线程来处理）
        consumer.registerMessageListener(new MessageListenerOrderly() {

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    log.info("消息内容: {}", new String(msg.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

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

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

    }
}

日志输出

生成者第1次启动发送消息，消费者输出：

在这里插入图片描述

生成者第2次启动发送消息，消费者输出：

在这里插入图片描述

原理参考：

RocketMQ的顺序消息（顺序消费）- 刘Java

RocketMQ的顺序消费问题MessageListenerOrderly()

4.3 延时消息

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

4.3.1 启动消息消费者

public class Consumer {

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

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.订阅主题Topic和Tag
        consumer.subscribe("DelayTopic", "*");

        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");

        //4.设置回调函数，处理消息
        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()) + "当前时间: " + sdf.format(new Date()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

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

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

4.3.2 发送延时消息

public class Producer {

    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {

        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");

        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.启动producer
        producer.start();

        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");


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

            //4.创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("DelayTopic", "Tag1", (sdf.format(new Date())).getBytes());

            //设定延迟时间（这里设置延迟时间的级别，2表示5s，可参考：MessageStoreConfig类）
            msg.setDelayTimeLevel(2);

            //5.发送消息
            SendResult result = producer.send(msg);

            //发送状态
            SendStatus status = result.getSendStatus();

            System.out.println("发送结果:" + result);

            //线程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }

        //6.关闭生产者producer
        producer.shutdown();
    }

}

###4.3.3 验证

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

消费者启动
消息ID：【C0A886056F9818B4AAC25C502A400000】  消息内容: 22:12:36当前时间: 22:12:43
消息ID：【C0A886056F9818B4AAC25C502E520001】  消息内容: 22:12:38当前时间: 22:12:43
消息ID：【C0A886056F9818B4AAC25C50324D0002】  消息内容: 22:12:39当前时间: 22:12:44
消息ID：【C0A886056F9818B4AAC25C5036440003】  消息内容: 22:12:40当前时间: 22:12:45
消息ID：【C0A886056F9818B4AAC25C503A2F0004】  消息内容: 22:12:41当前时间: 22:12:46
消息ID：【C0A886056F9818B4AAC25C503E1C0005】  消息内容: 22:12:42当前时间: 22:12:47
消息ID：【C0A886056F9818B4AAC25C5042080006】  消息内容: 22:12:43当前时间: 22:12:48
消息ID：【C0A886056F9818B4AAC25C5045F20007】  消息内容: 22:12:44当前时间: 22:12:49
消息ID：【C0A886056F9818B4AAC25C5049DD0008】  消息内容: 22:12:45当前时间: 22:12:50
消息ID：【C0A886056F9818B4AAC25C504DC90009】  消息内容: 22:12:46当前时间: 22:12:51

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

示例 demo

Consumer

public class Consumer {
    public static void main(String[] args) throws Exception {

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.订阅主题Topic和Tag
        consumer.subscribe("BatchTopic", "*");

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" 
                                       + Thread.currentThread().getName() + "," 
                                       + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

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

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

Producer

public class Producer {

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

        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");

        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.启动producer
        producer.start();


        List<Message> msgs = new ArrayList<Message>();


        //4.创建消息对象，指定主题Topic、Tag和消息体
        /**
         * 参数一：消息主题Topic
         * 参数二：消息Tag
         * 参数三：消息内容
         */
        Message msg1 = new Message("BatchTopic", "Tag1", 
                                   ("Hello World" + 1).getBytes());
        Message msg2 = new Message("BatchTopic", "Tag1", 
                                   ("Hello World" + 2).getBytes());
        Message msg3 = new Message("BatchTopic", "Tag1", 
                                   ("Hello World" + 3).getBytes());

        msgs.add(msg1);
        msgs.add(msg2);
        msgs.add(msg3);

        //5.发送消息
        SendResult result = producer.send(msgs);
        
        //发送状态
        SendStatus status = result.getSendStatus();

        System.out.println("发送结果:" + result);

        //线程睡1秒
        TimeUnit.SECONDS.sleep(1);

        //6.关闭生产者producer
        producer.shutdown();
    }

}

测试

可以看到对于生产者批量发送过来的消息，消费者成功如下消费了

消费者启动
consumeThread=ConsumeMessageThread_1,Hello World1
consumeThread=ConsumeMessageThread_2,Hello World2
consumeThread=ConsumeMessageThread_3,Hello World3

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

Tag 过滤示例 Demo

消费者使用 tag 作为过滤条件，只消费符合 Tag 过滤条件的消息

Producer

public class Producer {

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

        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");

        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.启动producer
        producer.start();

        for (int i = 1; i <= 3; i++) {

            //4.创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("FilterTagTopic", 
                                      "Tag" + i, 
                                      ("Hello World" + i).getBytes()
                                     );

            //5.发送消息
            SendResult result = producer.send(msg);

            //发送状态
            SendStatus status = result.getSendStatus();

            System.out.println("发送结果:" + result);

            //线程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }

        //6.关闭生产者producer
        producer.shutdown();
    }

}

Consumer

public class Consumer {
    public static void main(String[] args) throws Exception {

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.订阅主题Topic和Tag
        consumer.subscribe("FilterTagTopic", "Tag1 || Tag3 ");

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" 
                                       + Thread.currentThread().getName() + "," 
                                       + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

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

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

测试

发现只有 Tag1、Tag3 的消息被消费者消费了

消费者启动
consumeThread=ConsumeMessageThread_1,Hello World1
consumeThread=ConsumeMessageThread_2,Hello World3

SQL 语法过滤示例 Demo

问题

启动消费者时，遇到报错问题：The broker does not support consumer to filter message by SQL92，因此，需要在 broker-a.properties、broker-b-s.properties、broker-b.properties、broker-a-s.properties 配置文件中都得加上这个配置：enablePropertyFilter = true，然后，重启所有的 broker

Producer

public class Producer {

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

        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");

        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.启动producer
        producer.start();

        for (int i = 0; i < 10; i++) {
            
            //4.创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("FilterSQLTopic", 
                                      "Tag1", 
                                      ("Hello World" + i).getBytes());

            // 添加用户属性, 在消费者中可用作为sql过滤条件
            msg.putUserProperty("i", String.valueOf(i));

            //5.发送消息
            SendResult result = producer.send(msg);

            //发送状态
            SendStatus status = result.getSendStatus();

            System.out.println("发送结果:" + result);

            //线程睡1秒
            TimeUnit.SECONDS.sleep(2);
        }

        //6.关闭生产者producer
        producer.shutdown();
    }

}

Consumer

public class Consumer {

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

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.订阅主题Topic和Tag
        // 设置sql过滤条件 i>5
        consumer.subscribe("FilterSQLTopic", MessageSelector.bySql("i>5"));

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" 
                                       + Thread.currentThread().getName() + "," 
                                       + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

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

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

测试

一共发了10条消息（i从 0 开始），只接收到了5条i大于5的消息

消费者启动
consumeThread=ConsumeMessageThread_1,Hello World6
consumeThread=ConsumeMessageThread_2,Hello World7
consumeThread=ConsumeMessageThread_3,Hello World8
consumeThread=ConsumeMessageThread_4,Hello World9

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 Exception {
        
        //创建事务监听器
        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 {

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

4.6.2 使用限制

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

事务消息示例 Demo

Producer

/**
 * 发送同步消息
 */
public class Producer {

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

        //1.创建消息生产者producer，并制定生产者组名
        TransactionMQProducer producer = new TransactionMQProducer("group5");

        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //添加事务监听器
        producer.setTransactionListener(new TransactionListener() {

            /**
             * 在该方法中执行本地事务
             * @param msg
             * @param arg
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, 
                                                                 Object arg) {

                if (StringUtils.equals("TAGA", msg.getTags())) {

                    // 提交
                    return LocalTransactionState.COMMIT_MESSAGE;

                } else if (StringUtils.equals("TAGB", msg.getTags())) {

                    // 回滚
                    return LocalTransactionState.ROLLBACK_MESSAGE;

                } else if (StringUtils.equals("TAGC", msg.getTags())) {

                    // 返回此状态, 将触发MQ对该事务消息对应的事务的状态的回查
                    return LocalTransactionState.UNKNOW;
                }

                return LocalTransactionState.UNKNOW;
            }

            /**
             * 该方法是MQ进行消息事务状态回查
             * @param msg
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {

                System.out.println("消息的Tag:" + msg.getTags());

                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        //3.启动producer
        producer.start();

        String[] tags = {"TAGA", "TAGB", "TAGC"};

        for (int i = 0; i < 3; i++) {

            //4.创建消息对象，指定主题Topic、Tag和消息体

            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("TransactionTopic",
                                      tags[i],
                                      ("Hello World" + i).getBytes());

            //5.发送消息
            SendResult result = producer.sendMessageInTransaction(msg, null);

            //发送状态
            SendStatus status = result.getSendStatus();

            System.out.println("发送结果:" + result);

            //线程睡1秒
            TimeUnit.SECONDS.sleep(2);
        }

        //6.不要关闭生产者producer
        //producer.shutdown();
    }
}

Consumer

/**
 * 消息的接受者
 */
public class Consumer {

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

        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.134.3:9876;192.168.134.4:9876");

        //3.订阅主题Topic和Tag
        consumer.subscribe("TransactionTopic", "*");


        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {

                for (MessageExt msg : msgs) {
                    
                    System.out.println("consumeThread=" 
                                       + Thread.currentThread().getName() 
                                       + ","
                                       + msg.getTags() + "~"
                                       + new String(msg.getBody())
                                      );
                 }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

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

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

测试

从以下输出，可以看到消费者仅消费到了 TAGA 和 TAGC 消息，TAGC 被回查，TAGB 被回滚因而被删

生产者控制台输出

发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A88605E37418B4AAC25F15FC250000, offsetMsgId=null, messageQueue=MessageQueue [topic=TransactionTopic, brokerName=broker-b, queueId=2], queueOffset=0]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A88605E37418B4AAC25F1604090001, offsetMsgId=null, messageQueue=MessageQueue [topic=TransactionTopic, brokerName=broker-b, queueId=3], queueOffset=1]
发送结果:SendResult [sendStatus=SEND_OK, msgId=C0A88605E37418B4AAC25F160BE80002, offsetMsgId=null, messageQueue=MessageQueue [topic=TransactionTopic, brokerName=broker-a, queueId=0], queueOffset=0]
消息的Tag:TAGC

消费者控制台输出

消费者者启动
consumeThread=ConsumeMessageThread_1,TAGA~Hello World0
consumeThread=ConsumeMessageThread_2,TAGC~Hello World2