mq的一些前置知识和概念知识可以看这篇文章——SpringCloud入门(3) RabbitMQ，比如常见mq的对比等等，这篇文章不再赘述。

RocketMQ概念、安装与配置

Producer：消息的发送者，生产者；举例：发件人
Consumer：消息接收者，消费者；举例：收件人
Broker：暂存和传输消息的通道；举例：快递
NameServer：管理Broker；举例：各个快递公司的管理机构 相当于broker的注册中心，保留了broker的信息
Queue：队列，消息存放的位置，一个Broker中可以有多个队列
Topic：主题，消息的分类
ProducerGroup：生产者组
ConsumerGroup：消费者组，多个消费者组可以同时消费一个主题的消息，同一个组内的消费者订阅关系必须一致。一份消息会传递给每个组，可以被多个消费者组消费，至于组内是广播还是定向则可以自己配置。

消息发送的流程是，Producer询问NameServer，NameServer分配一个broker 然后Consumer也要询问NameServer，得到一个具体的broker，然后消费消息

了解了mq的基本概念和角色以后，我们开始安装rocketmq，建议在linux上.下载地址

注意选择版本，这里我们选择4.9.2的版本，后面使用alibaba时对应。source是源码版本，也可以下载下来学习。

然后上传服务器在root目录下创建文件夹mkdir rocketmq,将下载后的压缩包上传到阿里云服务器或者虚拟机中去.解压unzip rocketmq-all-4.9.2-bin-release.zip解压，如果你的服务器没有unzip命令，则下载安装一个yum install unzip

Benchmark：包含一些性能测试的脚本；Bin：可执行文件目录；Conf：配置文件目录；Lib：第三方依赖LICENSE：授权信息;NOTICE：版本公告；

然后配置环境变量vim /etc/profile,在文件末尾添加export NAMESRV_ADDR=阿里云公网IP:9876，如果你是云服务器比如阿里云你要写公网地址。比如我这里用的虚拟机，可以填localhost:9876 修改完之后source /etc/profile刷新一下

然后进入bin目录下，观察mqnamesrv启动项运行文件可知，启动项最终真正运行的是runserver.sh文件：

再进入到runserver.sh，我们发现如果使用的是JDK8，配置的JVM要4个G的内存，最大内存可到8个G，这样我们用于学习的虚拟机或者云服务器够呛能顶住。

所以我们需要修改一下配置，也用不到那么多的内存.修改runserver.sh文件,将71行和76行的Xms和Xmx等改小一点。vim runserver.sh,:set nu可以让vim看到行号，修改后:wq保存退出。没学过vim的同学建议学一下，后端必备的知识。

同理，观察mqbroker文件也是一样，启动的是runbroker.sh，我们修改67行，修改内存：

最后还要修改broker的配置文件，进入conf目录下，修改broker.conf文件

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
#添加如下字段
namesrvAddr=localhost:9876
autoCreateTopicEnable=true
brokerIP1=阿里云公网IP

添加参数解释
namesrvAddr：nameSrv地址 可以写localhost nameSrv和broker可以在一个服务器也可以不在
autoCreateTopicEnable：发送消息时如果没有这个Topic则自动创建主题，不然需要手动创建出来，很舒服
brokerIP1：broker也需要一个公网ip，如果不指定，那么是云服务器的内网地址，我们在本地无法连接使用。注意这里不能写成localhost！因为broker把自己的地址登记到name server上，等到发送方去name server找的时候如果写localhost，那肯定连不上。

完成配置之后我们就可以启动了，首先在安装目录下创建一个logs文件夹，mkdir logs,用于存放日志

先启动nameserver，如果直接./mqnamesrv，这个是前台运行，但我们想后台运行，在java里我们是怎样后台运行呢？我们是nohup java -jar xxx.jar &,我们这里也差不多，在bin目录下运行nohup sh mqnamesrv > ../logs/namesrv.log & ，>后面指定日志输出的位置

运行完之后怎么看呢？java里是怎么看？ps -ef |grep java,但是这样太麻烦了，可以直接使用jps命令查看java进程，只要安装了jdk就有这个命令，就可以看到本地虚拟机唯一ID(pid即进程号)和Java虚拟机进程的执行主类。这样看的可能不详细，我们使用jps -l，可以直接看到它所有的包名：

一定要先启动nameserver再启动broker。启动broker，也是一样的，nohup sh mqbroker -c ../conf/broker.conf > ../logs/broker.log & 。-c后面跟配置文件,>后面指定日志输出的位置。

启动之后我们还想要一个可视化窗口来看，Rocketmq 控制台可以可视化MQ的消息发送，旧版本源码是在rocketmq-external里的rocketmq-console，新版本已经单独拆分成，dashboard，下载地址。

下载后解压出来，在跟目录下执行mvn clean package -Dmaven.test.skip=true,在target目录下得到rocketmq-dashboard-1.0.0.jar，将jar包上传到服务器上去，然后运行nohup java -jar rocketmq-dashboard-1.0.0.jar --server.port=8001 --rocketmq.config.namesrvAddr=127.0.0.1:9876 > ./rocketmq-4.9.2/logs/dashboard.log &

其中--server.port我们指定运行的端口，否则默认是8080
--rocketmq.config.namesrvAddr=127.0.0.1:9876 指定namesrv地址
然后访问IP(我这里是用虚拟机所以是虚拟机地址):8001

注意如果启动失败的话，要学会去看日志，别忘记我们在启动nameserver、broker还有dashboard时都指定了生成的日志路径，日志里的记录大部分情况下足够dubug了。

Tips:还有一些可能会用到的命令
jps -l查看到进程号之后 如果想重启服务 可以使用kill 进程号来请求终止进程
因为nameserver、broker还有dashboard都很占内存，所以通过free -mh或命令查看一下内存使用情况，如果total-used，即剩余内存大于20%就是安全的。

docker配置

上面的安装流程只是为了入门，真正实际开发大部分还是docker部署，所以这里在贴一下docker部署的过程，docker学习可以参考这篇博客。

下载RockerMQ需要的镜像：docker pull rocketmqinc/rocketmq、docker pull styletang/rocketmq-console-ng

创建NameServer数据存储路径mkdir -p /home/rocketmq/data/namesrv/logs /home/rocketmq/data/namesrv/store

启动NameServer容器docker run -d --name rmqnamesrv -p 9876:9876 -v /home/rocketmq/data/namesrv/logs:/root/logs -v /home/rocketmq/data/namesrv/store:/root/store -e "MAX_POSSIBLE_HEAP=100000000" rocketmqinc/rocketmq sh mqnamesrv

创建Broker数据存储路径mkdir -p /home/rocketmq/data/broker/logs /home/rocketmq/data/broker/store

创建conf配置文件目录mkdir /home/rocketmq/conf

在配置文件目录下创建broker.conf配置文件

# 所属集群名称，如果节点较多可以配置多个
brokerClusterName = DefaultCluster
#broker名称，master和slave使用相同的名称，表明他们的主从关系
brokerName = broker-a
#0表示Master，大于0表示不同的slave
brokerId = 0
#表示几点做消息删除动作，默认是凌晨4点
deleteWhen = 04
#在磁盘上保留消息的时长，单位是小时
fileReservedTime = 48
#有三个值：SYNC_MASTER，ASYNC_MASTER，SLAVE；同步和异步表示Master和Slave之间同步数据的机制；
brokerRole = ASYNC_MASTER
#刷盘策略，取值为：ASYNC_FLUSH，SYNC_FLUSH表示同步刷盘和异步刷盘；SYNC_FLUSH消息写入磁盘后才返回成功状态，ASYNC_FLUSH不需要；
flushDiskType = ASYNC_FLUSH
# 设置broker节点所在服务器的ip地址
brokerIP1 = 你服务器外网ip

启动Broker容器docker run -d --name rmqbroker --link rmqnamesrv:namesrv -p 10911:10911 -p 10909:10909 -v /home/rocketmq/data/broker/logs:/root/logs -v /home/rocketmq/data/broker/store:/root/store -v /home/rocketmq/conf/broker.conf:/opt/rocketmq-4.4.0/conf/broker.conf --privileged=true -e "NAMESRV_ADDR=namesrv:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c /opt/rocketmq-4.4.0/conf/broker.conf

启动控制台docker run -d --name rmqadmin -e "JAVA_OPTS=-Drocketmq.namesrv.addr=你的外网地址:9876 \ -Dcom.rocketmq.sendMessageWithVIPChannel=false \ -Duser.timezone='Asia/Shanghai'" -v /etc/localtime:/etc/localtime -p 9999:8080 styletang/rocketmq-console-ng

正常启动后的docker ps:

访问控制台，你的服务器外网ip:9999

RocketMQ快速入门

RocketMQ提供了发送多种发送消息的模式，例如同步消息，异步消息，顺序消息，延迟消息，事务消息等，我们一一学习。但是我们先搞清楚消息发送和监听的流程，然后我们在开始敲代码

消息生产者：

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

消息消费者:

1.创建消费者consumer，制定消费者组名
2.指定Nameserver地址
3.创建监听订阅主题Topic和Tag等
4.处理消息
5.启动消费者consumer

**同步消息

发送同步消息，发送过后会有一个返回值，也就是mq服务器接收到消息后返回的一个确认，生产者在发送后会等待mq主机的返回确认，这种方式非常安全，但是性能上并没有这么高，而且在mq集群中，也是要等到所有的从机都复制了消息以后才会返回，所以针对重要的消息可以选择这种方式，因为几乎任何的mq都会存在消息丢失的风险。

搭建一个Rocketmq-demo，引入依赖：

<!-- 原生的api   -->
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.9.2</version><!--docker的用下面这个版本--><version>4.4.0</version>
</dependency>
<dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version>
</dependency>
<dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional>
</dependency>

编写生产者

@Test
public void simpleProducer() throws Exception {// 创建一个生产者 使用建默认的生产者  (制定一个组名)DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");// 连接namesrv 设置nameServer地址producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 字符常量 ip:9876// 启动producer.start();// 创建一个消息for (int i = 0; i < 10; i++) {// 创建消息 第一个参数：主题的名字 第二个参数：消息内容Message message = new Message("testTopic", "我是一个简单的消息".getBytes());// 发送消息 有返回值 可以打印一下SendResult sendResult = producer.send(message);System.out.println(sendResult.getSendStatus());}// 关闭生产者producer.shutdown();
}

编写消费者

 消费者
@Test
public void simpleConsumer() throws Exception {// 创建一个消费者 默认消费者组  (制定一个组名)DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");// 连接namesrv 设置nameServer地址consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 字符常量 ip:9876// 订阅一个主题  写入主题名和消息表达式 *标识订阅这个主题中所有的消息  后期再介绍消息过滤consumer.subscribe("testTopic", "*");// 设置一个监听器 (一直监听的， 异步回调方式，消费线程和主线程不是一个线程)// MessageListenerConcurrently 是多线程消费，默认20个线程，可以通过 consumer.setConsumeThreadMax() 来设置最大消费线程数consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {// 这个就是消费的方法 （业务处理）System.out.println("我是消费者");// 这里的消息虽然是个List 但默认只有一个消息 后面有配置批量消息这样的概念System.out.println(msgs.get(0).toString());System.out.println("消息内容:" + new String(msgs.get(0).getBody()));System.out.println("消费上下文:" + context);// 返回消费的状态 如果是CONSUME_SUCCESS 则成功，若为RECONSUME_LATER则该条消息会被重回队列，重新被投递// 重试的时间为messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h// 也就是第一次1s 第二次5s 第三次10s .... 如果重试了16次 那么这个消息就会被终止发送给消费者return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动 这个start一定要写在registerMessageListener下面consumer.start();// 主线程不能退出，否则消费者线程也会随之退出 挂起当前 JVM，保持主线程存活，以便消费者能持续运行System.in.read(); // 让程序在该行停下来，等待用户输入
}

注意MessageListenerConcurrently 是多线程消费，默认20个线程，可以通过 consumer.setConsumeThreadMax() 来设置最大消费线程数。所以当有消息同时进来了，消息的处理是并行的。

测试

发送后消息，输出如下，rocketmq控制面板的主题多了一个我们刚刚发送的，我们再来详细看一下主题里的这些按钮的一些功能。

查看其状态可知，一个broker可以有不同的topic，我们这里只配了一个broker，一个topic默认有4个队列，采用默认的负载均衡算法往里面去放消息，看如下图可知，消息分配是很均匀的。

这里补充一点知识，如果是广播模式，那好说，一个组内所有的消费者都会拿到每个队列里的消息，如果是负载均衡模式，假设有c1、c2两个消费者，那么每个队列是要固定联系好消费者的，即比如队列1、2的消息只会给c1，2、3的消息只会给c2，比如4个队列都指定了消费者，组内再有一个消费者，那它永远没有消息。
所以最好队列数量>=组内消费者数量

然后去看它的路由，主要是broker的地址，先通过nameserver找到broker，然后通过broker连接发送的消息。这里还有读队列写队列还有权限，先不管后面再介绍。

点击consumer管理可以发现现在还没有订阅者，

topic配置可以修改一些配置：

也可以通过面板发送消息，但一般不这样做。

还有重置消费位点，跳过堆积，删除这三个操作，我们后面介绍。

测试消费代码：
中间就是消息的内容，里面有很多东西，broker的名字，队列号，队列偏移量，还有一些状态、时间，消息内容在body里以字节数组存放着

此时ui控制台的消费者管理就能看到消费者了，消费者都是一个人，这里的代理者位点、消费者位点、差值是什么意思呢？
mq是代理者，我们的程序就是消费者，说白了就是4个队列中的消息，总共接收、已消费、未消费的消息数量。每消息成功消息一次，消费者位点就移动增加一位，队列每接收到一条消息，代理者位点就增加一位。

消费模式

MQ的消费模式可以大致分为两种，一种是推Push，一种是拉Pull。我们刚刚写的代码都是Push–>DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group"); 现在官方不推荐使用Pull了–>DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("test-consumer-group");

官方推荐不再使用Pull API的原因
复杂性和易用性： Pull模式需要用户手动管理拉取逻辑，包括拉取频率、拉取数量、异常处理、消息重试等。这增加了实现的复杂性，不容易使用和维护。相反，Push模式通过封装这些复杂性，提供了更高层次的API，简化了开发工作。
性能和资源管理： 在Push模式中，RocketMQ客户端能够更好地管理和优化拉取逻辑，包括流控和负载均衡，从而提高整体性能和资源利用率。手动实现这些优化在Pull模式中是比较困难的。
一致性和可靠性： Push模式内置了很多一致性和可靠性的保障机制，例如消费确认、重试机制等。这些在Pull模式中需要用户自己实现，容易出错。

• Push是服务端【MQ】主动推送消息给客户端，优点是及时性较好，但如果客户端没有做好流控，一旦服务端推送大量消息到客户端时，就会导致客户端消息堆积甚至崩溃。

• Pull是客户端需要主动到服务端取数据，优点是客户端可以依据自己的消费能力进行消费，但拉取的频率也需要用户自己控制，拉取频繁容易造成服务端和客户端的压力，拉取间隔长又容易造成消费不及时。

其实Push模式也是基于Pull模式的，只是客户端内部封装了api，通过长轮询方式实现的，使得消费者不需要手动拉取，而是通过回调或者监听的方式获取消息。因为任何mq中间件都是Pull模式，都需要你主动去拉，它不会主动推给你，mq的qps很高，一有消息全部都主动推给你服务端肯定受不了的。

总结：一般场景下，上游消息生产量小或者均速的时候，可以选择Push模式。在特殊场景下，例如电商大促，抢优惠券等场景可以选择Pull模式。

**异步消息

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

编写生产者

@Test
public void asyncProducer() throws Exception {// 创建默认的生产者DefaultMQProducer producer = new DefaultMQProducer("async-producer-group");producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 设置nameServer地址producer.start(); // 启动实例Message message = new Message("asyncTopic", "我是一个异步消息".getBytes());producer.send(message, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("发送成功");}@Overridepublic void onException(Throwable e) {System.err.println("发送失败:" + e.getMessage());}});System.out.println("我先执行");System.in.read(); // 挂起jvm 因为回调是异步的不然测试不出来 键入回车键就解除挂起producer.shutdown(); // 关闭实例
}

消费者的代码可与同步消息使用同一个。

*单向消息

这种方式主要用在不关心发送结果的场景，这种方式吞吐量很大，但是存在消息丢失的风险，例如日志信息的发送

生产者代码

@Test
public void onewayProducer() throws Exception {// 创建默认的生产者DefaultMQProducer producer = new DefaultMQProducer("oneway-producer-group");producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 创建默认的生产者producer.start(); // 启动实例Message message = new Message("onewayTopic", "日志xxx".getBytes());producer.sendOneway(message); // 发送单向消息System.out.println("成功");producer.shutdown(); // 关闭实例
}

消费者代码同上，测试的时候注意主题topic更换

拓充一下日志采集的思路
开发环境，直接输出在控制台即可
生产环境下，可以输出在文件，但是查看起来就很麻烦了，那么可以输出到mysql，但是存储的内容有限，后期我们还会学到ES，存储的数据量级就很大了、
再开扩一下思路，每个业务操作都要记录日志，虽然麻烦但是要有，那么每个操作都有一个插入日志到mysql或者es的操作不就很费时了吗？那么可以专门为日志操作搭建一个服务器log-service，业务操作是生产者，再有一个消费者来记录日志，mq是异步的，这样性能也好很多。

**延迟消息

消息放入mq后，过一段时间，才会被监听到，然后消费
比如下订单业务，提交了一个订单就可以发送一个延时消息，30min后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

生产者代码

@Test
public void msProducer() throws Exception {// 创建默认的生产者DefaultMQProducer producer = new DefaultMQProducer("ms-producer-group");producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 设置nameServer地址producer.start(); // 启动实例Message message = new Message("orderMsTopic", "订单号，座位号".getBytes());// 给消息设置一个延迟时间// messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"message.setDelayTimeLevel(3); // 10s延迟// 发延迟消息 这里没有获取返回值 所以是单向的producer.send(message);System.out.println("发送时间" + new Date()); // 打印时间producer.shutdown(); // 关闭实例
}

消费者代码同上，可以自己改一下测试一下，在消费方法里打印一下当前时间，测试发现其实不是很精确的10s，与官方的时间还是有误差的，误差还是很大，我自己的环境测试26s才接收到，因为broker默认是8个G的内存，我们给修改成256m了，性能不足。第一次测试比较慢，第二次就比较准了，因为第一次还需要加载一些东西，冷加载嘛。

发送时间Mon May 27 16:36:24 CST 2024收到消息时间:Mon May 27 16:36:50 CST 2024
订单号，座位号# 第二次测试发送时间Mon May 27 16:43:56 CST 2024收到消息时间:Mon May 27 16:44:06 CST 2024
订单号，座位号

*顺序消息

消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为：分区有序或者全局有序。
可能大家会有疑问，mq不就是FIFO吗？
rocketMq的broker的机制，导致了rocketMq会有这个问题
因为一个broker中对应了四个queue

在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候以多线程的方式从多个queue上拉取消息，这种情况发送和消费是不能保证顺序，所以肯定要修改为以单线程的方式去消费。

如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；但也没必要全局有序，我们只需要确保局部有序就行了，如果多个queue参与，各个queue内的消息是相对有序的，则为分区有序，即相对每个queue，消息都是有序的。

下面用订单进行分区有序的示例。一个订单的顺序流程是：下订单、发短信通知、物流、签收，必须依次执行。订单顺序号相同的消息会被先后发送到同一个队列中，消费时，同一个顺序获取到的肯定是同一个队列。

模拟一个订单的发送流程，创建两个订单，发送的消息分别是
订单号111 消息流程 下订单->物流->签收
订单号112 消息流程 下订单->物流->拒收

先创建一个订单对象

@Data
@AllArgsConstructor
@NoArgsConstructor
public class MsgModel {private String orderSn;private Integer userId;private String desc; // 下单 短信 物流// xxx
}

生产者代码

其中发送消息有个很重要的api，用于消息的选择，我们ctrl+P可以查看：

生产者代码

@Test
public void testOrderlyProducer() throws Exception {// 创建默认的生产者DefaultMQProducer producer = new DefaultMQProducer("orderly-producer-group");producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 设置nameServer地址producer.start(); // 启动实例List<Order> orderList = Arrays.asList(new Order(1, 111, 59D, new Date(), "下订单"),new Order(2, 111, 59D, new Date(), "物流"),new Order(3, 111, 59D, new Date(), "签收"),new Order(4, 112, 89D, new Date(), "下订单"),new Order(5, 112, 89D, new Date(), "物流"),new Order(6, 112, 89D, new Date(), "拒收"));// 循环集合开始发送 发送顺序消息发送时要确保有序 并且要发到同一个队列下面去orderList.forEach(order -> {Message message = new Message("orderlyTopic", order.toString().getBytes());try {// 发 相同的订单号去相同的队列producer.send(message, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { // 在这里 选择队列// 这里重新函数的mqs就是所有的队列(默认是4个) msg就是这条消息 arg是send函数的第三个参数传入进来的int queueNumber = mqs.size(); // 当前主题有多少个队列// 这个arg就是后面传入的 order.getOrderNumber()Integer i = (Integer) arg;int index = i % queueNumber; // 用这个值去%队列的个数得到一个队列// 返回选择的这个队列即可 ，那么相同的订单号 就会被放在相同的队列里 实现FIFO了return mqs.get(index); // 选择发送到第几个队列}}, order.getOrderNumber()); // msgModel.getOrderSn()是send函数的第三个参数 这个值传给了select函数的arg} catch (Exception e) {System.out.println("发送异常");}});producer.shutdown(); // 关闭实例System.out.println("发送完成");
}

发送完可以看到消息被均匀的放在了两个队列里：

消费者代码，消费者也有讲究，也是刚刚提到的，不能再用并发模式了，而是采用单线程去消费，或者采用多线程，将最大线程数设置为1也行。
注意这里的单线程不是全局只有一个，是针对每个队列单独使用一个线程进行顺序消费，而不是整个消费者实例只有一个线程，我们可以打印一下线程id来看一下。

@Test
public void orderlyConsumer() throws Exception {// 创建默认消费者组DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderly-consumer-group");consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 设置nameServer地址// 订阅一个主题来消费 *表示没有过滤参数 表示这个主题的任何消息consumer.subscribe("orderlyTopic", "*");// MessageListenerConcurrently 并发模式 多线程的  重试16次// MessageListenerOrderly 顺序模式 单线程的 无限重试Integer.Max_Valueconsumer.registerMessageListener(new MessageListenerOrderly() {@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {System.out.println("线程id:" + Thread.currentThread().getId());System.out.println(new String(msgs.get(0).getBody()));return ConsumeOrderlyStatus.SUCCESS;}});consumer.start();System.in.read();
}

我的这个运行有点太理想了，多尝试几次可以发现，有可能出现队列之间交叉消费，但是不管怎样，队内的顺序是相对有序的。

针对ConsumeOrderlyStatus，它跟ConsumeConcurrentlyStatus还是有一些区别，我们看一下：

然后消费者代码里这个SUSPEND_CURRENT_QUEUE_A_MOMENT意思就是消费失败了，队列里有顺序消息A B C，A消费失败了，不会去消费B和C，而是挂起当前的A消息，等一会再去消费，如果A一直报错，则会一直重试，直到A成功才去消费B，

关于重试时间和次数，可以去查看一下DefaultMQPushConsumer的源码，默认重试间隔为1s，重试次数无限制。可以自己设置每次重试的时间间隔（单位：毫秒) consumer.setSuspendCurrentQueueTimeMillis(2000);// 2秒

设置最大重试次数 consumer.setMaxReconsumeTimes(5); // 最大重试5次

批量消息

Rocketmq可以一次性发送一组消息，那么这一组消息会被当做一个消息消费

这种方式发送的消息，可以测试一下，在前端仪表板看看主题状态，发送的这些打包的消息都是放在一个队列里的，但是消息还是有三个的，等于说只有发是打包的，可以测试一下消费端，输出一下接收的消息的长度，还是只有一个。

生产者代码

@Test
public void testBatchProducer() throws Exception {// 创建默认的生产者DefaultMQProducer producer = new DefaultMQProducer("batch-producer-group");// 设置nameServer地址producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);// 启动实例producer.start();List<Message> msgs = Arrays.asList(new Message("batchTopic", "我是一组消息的A消息".getBytes()),new Message("batchTopic", "我是一组消息的B消息".getBytes()),new Message("batchTopic", "我是一组消息的C消息".getBytes()));SendResult send = producer.send(msgs);System.out.println(send);// 关闭实例producer.shutdown();
}

消费者代码

@Test
public void msConsumer() throws Exception {// 创建默认消费者组DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch-consumer-group");// 设置nameServer地址consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);// 订阅一个主题来消费和订阅消息表达式，默认是*consumer.subscribe("batchTopic", "*");// 注册一个消费监听 MessageListenerConcurrently是并发消费// 默认是20个线程一起消费，可以通过 consumer.setConsumeThreadMax() 来设置consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {// 这里执行消费的代码 默认是多线程消费System.out.println(Thread.currentThread().getName() + "----" + new String(msgs.get(0).getBody()));System.out.println("收到消息了" + new Date());System.out.println(msgs.size());return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();
}

事务消息

这个可以先不学了，比较鸡肋，基本不用，后面我们会专门学分布式事务seata，它是专门解决分布式事务问题的。这里也附上代码，有兴趣可以学一下。

事务消息的发送流程。它可以被认为是一个两阶段的提交消息实现，以确保分布式系统的最终一致性。事务性消息确保本地事务的执行和消息的发送可以原子地执行。

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

事务消息发送及提交

1. 发送消息（half消息）。
2. 服务端响应消息写入结果。
3. 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）。
4. 根据本地事务状态执行Commit或Rollback（Commit操作生成消息索引，消息对消费者可见）

事务补偿

1. 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”
2. Producer收到回查消息，检查回查消息对应的本地事务的状态
3. 根据本地事务状态，重新Commit或者Rollback
   其中，补偿阶段用于解决消息UNKNOW或者Rollback发生超时或者失败的情况。

事务消息状态

事务消息共有三种状态，提交状态、回滚状态、中间状态：
 TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
 TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
 TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。

事务消息生产者代码

/*** TransactionalMessageCheckService的检测频率默认1分钟，可通过在broker.conf文件中设置transactionCheckInterval的值来改变默认值，单位为毫秒。* 从broker配置文件中获取transactionTimeOut参数值。* 从broker配置文件中获取transactionCheckMax参数值，表示事务的最大检测次数，如果超过检测次数，消息会默认为丢弃，即回滚消息。** @throws Exception*/
@Test
public void testTransactionProducer() throws Exception {// 创建一个事务消息生产者TransactionMQProducer producer = new TransactionMQProducer("test-group");producer.setNamesrvAddr("localhost:9876");// 设置事务消息监听器producer.setTransactionListener(new TransactionListener() {// 这个是执行本地业务方法@Overridepublic LocalTransactionState executeLocalTransaction(Message msg, Object arg) {System.out.println(new Date());System.out.println(new String(msg.getBody()));// 这个可以使用try catch对业务代码进行性包裹// COMMIT_MESSAGE 表示允许消费者消费该消息// ROLLBACK_MESSAGE 表示该消息将被删除，不允许消费// UNKNOW表示需要MQ回查才能确定状态 那么过一会 代码会走下面的checkLocalTransaction(msg)方法return LocalTransactionState.UNKNOW;}// 这里是回查方法 回查不是再次执行业务操作，而是确认上面的操作是否有结果// 默认是1min回查 默认回查15次 超过次数则丢弃打印日志 可以通过参数设置// transactionTimeOut 超时时间// transactionCheckMax 最大回查次数// transactionCheckInterval 回查间隔时间单位毫秒// 触发条件// 1.当上面执行本地事务返回结果UNKNOW时,或者下面的回查方法也返回UNKNOW时 会触发回查// 2.当上面操作超过20s没有做出一个结果，也就是超时或者卡主了，也会进行回查@Overridepublic LocalTransactionState checkLocalTransaction(MessageExt msg) {System.err.println(new Date());System.err.println(new String(msg.getBody()));// 这里return LocalTransactionState.UNKNOW;}});producer.start();Message message = new Message("TopicTest2", "我是一个事务消息".getBytes());// 发送消息producer.sendMessageInTransaction(message, null);System.out.println(new Date());System.in.read();
}

事务消费者代码

@Test
public void testTransactionConsumer() throws Exception {// 创建默认消费者组DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");// 设置nameServer地址consumer.setNamesrvAddr("localhost:9876");// 订阅一个主题来消费 *表示没有过滤参数 表示这个主题的任何消息consumer.subscribe("TopicTest2", "*");// 注册一个消费监听 MessageListenerConcurrently是并发消费// 默认是20个线程一起消费，可以参看 consumer.setConsumeThreadMax()consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {// 这里执行消费的代码 默认是多线程消费System.out.println(Thread.currentThread().getName() + "----" + new String(msgs.get(0).getBody()));return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();
}

测试结果

*发送带标签的消息，消息过滤

Rocketmq提供消息过滤功能，通过tag或者key进行区分
我们往一个主题里面发送消息的时候，根据业务逻辑，可能需要区分，比如带有tagA标签的被A消费，带有tagB标签的被B消费，还有在事务监听的类里面，只要是事务消息都要走同一个监听，我们也需要通过过滤才区别对待

什么时候该用 Topic，什么时候该用 Tag？
总结：不同的业务应该使用不同的Topic如果是相同的业务里面有不同表的表现形式，那么我们要使用tag进行区分

可以从以下几个方面进行判断：
1.消息类型是否一致：如普通消息、事务消息、定时（延时）消息、顺序消息，不同的消息类型使用不同的 Topic，无法通过 Tag 进行区分。
2.业务是否相关联：没有直接关联的消息，如淘宝交易消息，京东物流消息使用不同的 Topic 进行区分；而同样是天猫交易消息，电器类订单、女装类订单、化妆品类订单的消息可以用 Tag 进行区分。
3.消息优先级是否一致：如同样是物流消息，盒马必须小时内送达，天猫超市 24 小时内送达，淘宝物流则相对会慢一些，不同优先级的消息用不同的 Topic 进行区分。
4.消息量级是否相当：有些业务消息虽然量小但是实时性要求高，如果跟某些万亿量级的消息使用同一个 Topic，则有可能会因为过长的等待时间而“饿死”，此时需要将不同量级的消息进行拆分，使用不同的 Topic。
总的来说，针对消息分类，您可以选择创建多个 Topic，或者在同一个 Topic 下创建多个 Tag。但通常情况下，不同的 Topic 之间的消息没有必然的联系，而 Tag 则用来区分同一个 Topic 下相互关联的消息，例如全集和子集的关系、流程先后的关系。

生产者代码

@Test
public void tagProducer() throws Exception {// 创建默认的生产者DefaultMQProducer producer = new DefaultMQProducer("tag-producer-group");producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 设置nameServer地址producer.start(); // 启动实例Message message = new Message("tagTopic", "vip1", "我是vip1的文章".getBytes());Message message2 = new Message("tagTopic", "vip2", "我是vip2的文章".getBytes());producer.send(message);producer.send(message2);System.out.println("发送成功");producer.shutdown(); // 关闭实例
}

消费者代码

/*** vip1** @throws Exception*/
@Test
public void tagConsumer1() throws Exception {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-a");consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);consumer.subscribe("tagTopic", "vip1");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {System.out.println("我是vip1的消费者，我正在消费消息" + new String(msgs.get(0).getBody()));return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();
}/*** vip1 || vip2** @throws Exception*/
@Test
public void tagConsumer2() throws Exception {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-b");consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);consumer.subscribe("tagTopic", "vip1 || vip2");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {System.out.println("我是vip2的消费者，我正在消费消息" + new String(msgs.get(0).getBody()));return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();
}

查看官方文档，可知订阅关系的定义：一个消费者组订阅一个Topic的某一个Tag，这种记录被称之为订阅关系。

注意！只有订阅关系完全一致才能称之为组，比如这里组a的订阅关系是主题tagTopic中的tag:vip1，组b的订阅关系是主题tagTopic中的tag:vip1||vip2，所以它两不能是同一个消费者组。如果订阅关系不一致，会导致消费消息紊乱，甚至消息丢失。

对于消费者组，我们之间提到：消费者组，多个消费者组可以同时消费一个主题的消息，同一个组内的消费者订阅关系必须一致。一份消息会传递给每个组，至于组内是广播还是定向则可以自己配置。

这部分逻辑强烈推荐阅读一下官方文档，其实这类技术还是学习官方文档上手更快。

所以测试这里的代码，vip1消息这两个消费者都能接收到，vip2消息则只有第二个消费者能接收到了。

想到如果消费者2隔了很久再去消费还能接收到这条消息吗，去查了相关资料：
RocketMQ的消息是持久化存储在Broker（消息代理）中的。默认情况下，RocketMQ会将消息存储7天，超过这个时间后消息会被自动删除。当然，这个存储时间是可以配置的，通过broker.conf中的参数fileReservedTime来进行设置。

**RocketMQ中消息的Key

在rocketmq中的消息，默认会有一个messageId当做消息的唯一标识，我们自己也可以给消息携带一个key，用作唯一标识或者业务标识，包括在控制面板查询的时候也可以使用messageId或者key来进行查询

生产者代码

/*** 业务参数 我们自身要确保唯一* 为了查阅和去重** @throws Exception*/
@Test
public void keyProducer() throws Exception {DefaultMQProducer producer = new DefaultMQProducer("key-producer-group");producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);producer.start();String key = UUID.randomUUID().toString(); // 通过UUID来唯一标识System.out.println(key);Message message = new Message("keyTopic", "vip1", key, "我是vip1的文章".getBytes());producer.send(message);System.out.println("发送成功");producer.shutdown();
}

消费者代码

@Test
public void keyConsumer() throws Exception {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("key-consumer-group");consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);consumer.subscribe("keyTopic", "*"); // key不影响订阅关系consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {MessageExt messageExt = msgs.get(0);System.out.println("我是vip1的消费者，我正在消费消息" + new String(messageExt.getBody()));System.out.println("我们业务的标识：" + messageExt.getKeys()); // key放在消息体里 可以拿到return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();
}

messageExt还可以拿到很多东西，可以自己去尝试一下。