RabbitMQ
2024/4/27大约 42 分钟
RabbitMQ介绍
几种常见MQ的对比：
| RabbitMQ | ActiveMQ | RocketMQ | Kafka | |
|---|---|---|---|---|
| 公司/社区 | Rabbit | Apache | 阿里 | Apache |
| 开发语言 | Erlang | Java | Java | Scala&Java |
| 协议支持 | AMQP，XMPP，SMTP，STOMP | OpenWire,STOMP，REST,XMPP,AMQP | 自定义协议 | 自定义协议 |
| 可用性 | 高 | 一般 | 高 | 高 |
| 单机吞吐量 | 一般 | 差 | 高 | 非常高 |
| 消息延迟 | 微秒级 | 毫秒级 | 毫秒级 | 毫秒以内 |
| 消息可靠性 | 高 | 一般 | 高 | 一般 |
追求可用性：Kafka、 RocketMQ 、RabbitMQ
追求可靠性：RabbitMQ、RocketMQ
追求吞吐能力：RocketMQ、Kafka
追求消息低延迟：RabbitMQ、Kafka
安装
RabbitMQ是基于Erlang语言开发的开源消息通信中间件
官网 RabbitMQ: One broker to queue them all | RabbitMQ
教程 RabbitMQ Tutorials | RabbitMQ
github RabbitMQ Tutorials | RabbitMQ
docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.13-managementdocker run \
-e RABBITMQ_DEFAULT_USER=cdd \
-e RABBITMQ_DEFAULT_PASS=1 \
-v mq-plugins:/plugins \
--name mq \
--hostname mq \
-p 15672:15672 \
-p 5672:5672 \
--network hmall \
-d \
rabbitmq:3.13-management
docker run \
-v mq-plugins:/plugins \
--name mq \
--hostname mq \
-p 15672:15672 \
-p 5672:5672 \
--network hmall \
-d \
rabbitmq:3.13-managementdocker run: Docker 命令，用于创建并运行一个容器。-e RABBITMQ_DEFAULT_USER=cdd: 设置环境变量RABBITMQ_DEFAULT_USER，指定 RabbitMQ 的默认用户名为cdd。-e RABBITMQ_DEFAULT_PASS=1: 设置环境变量RABBITMQ_DEFAULT_PASS，指定 RabbitMQ 的默认密码为1。-v mq-plugins:/plugins: 将本地卷mq-plugins挂载到容器内的/plugins目录，用于持久化 RabbitMQ 插件等数据。--name mq: 指定容器的名称为mq。--hostname mq: 指定容器的主机名为mq。-p 15672:15672: 将容器的15672端口映射到主机的15672端口，用于访问 RabbitMQ 的 Web 管理界面。-p 5672:5672: 将容器的5672端口映射到主机的5672端口，用于 RabbitMQ 客户端连接。--network hmall: 将容器连接到hmall网络，使得容器可以与同一网络中的其他容器进行通信。-d: 以守护进程方式运行容器，即在后台运行。rabbitmq:3.13-management: 指定要运行的镜像名称及版本号，这里是 RabbitMQ 的官方镜像版本3.13，并包含了管理界面。
图形化界面
默认用户名密码：guest
整体架构
publisher：生产者，也就是发送消息的一方consumer：消费者，也就是消费消息的一方queue：队列，存储消息。生产者投递的消息会暂存在消息队列中，等待消费者处理exchange：交换机，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。virtual host：虚拟主机，起到数据隔离的作用。每个虚拟主机相互独立，有各自的exchange、queue
发送消息先到交换机，交换机路由到队列，消费者监听队列
快速入门
控制台界面创建队列
交换机绑定队列
发送消息：
数据隔离
virtual host 虚拟主机
Java客户端
快速入门
AMQP
由于RabbitMQ采用了AMQP协议，因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息，都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。
但是，RabbitMQ官方提供的Java客户端编码相对复杂，一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具：SpringAMQP。并且还基于SpringBoot对其实现了自动装配，使用起来非常方便。
SpringAmqp的官方地址：
Spring AMQP
SpringAMQP提供了三个功能：
- 自动声明队列、交换机及其绑定关系
- 基于注解的监听器模式，异步接收消息
- 封装了RabbitTemplate工具，用于发送消息
导入Demo工程
将其复制到工作空间，然后用Idea打开，项目结构如图：
包括三部分：
- mq-demo：父工程，管理项目依赖
- publisher：消息的发送者
- consumer：消息的消费者
在mq-demo这个父工程中，已经配置好了SpringAMQP相关的依赖：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>cn.itcast.demo</groupId>
<artifactId>mq-demo</artifactId>
<version>1.0-SNAPSHOT</version>
<modules>
<module>publisher</module>
<module>consumer</module>
</modules>
<packaging>pom</packaging>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.12</version>
<relativePath/>
</parent>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<!--AMQP依赖，包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!--单元测试-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
</dependencies>
</project>因此，子工程中就可以直接使用SpringAMQP了。
代码示例
发送
直接向队列发送消息，跳过交换机
添加配置
image-20240427143800995 logging: pattern: dateformat: MM-dd HH:mm:ss:SSS spring: rabbitmq: host: 10.61.2.143 # 你的虚拟机IP port: 5672 # 端口 virtual-host: learn # 虚拟主机 username: learn # 用户名 password: 1 # 密码测试类
/** * @Author：CharmingDaiDai * @Project：mq-demo * @Date：2024/4/27 下午2:38 */ package com.itheima.publisher; import org.junit.jupiter.api.Test; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; @SpringBootTest public class SpringAmqpTest { @Autowired private RabbitTemplate rabbitTemplate; @Test public void testSimpleQueue() { // 队列名称 String queueName = "learn.queue1"; // 消息 String message = "hello, spring amqp!"; // 发送消息 rabbitTemplate.convertAndSend(queueName, message); } }
接收
配置文件也要改
package com.itheima.consumer.Listeners;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @Author：CharmingDaiDai
* @Project：mq-demo
* @Date：2024/4/27 下午2:59
*/
@Slf4j
@Component
public class MqListener {
@RabbitListener(queues = "learn.queue1")
public void listen(String message) {
System.out.println("message = " + message);
}
}
Work模型
Work queues，任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。
当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。
此时就可以使用work 模型，多个消费者共同处理消息，消息处理的速度就能大大提高了。
监听
package com.itheima.consumer.Listeners; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * @Author：CharmingDaiDai * @Project：mq-demo * @Date：2024/4/27 下午2:59 */ @Slf4j @Component public class MqListener { @RabbitListener(queues = "learn.queue1") public void listen(String message) { System.out.println("message = " + message); } @RabbitListener(queues = "work.queue1") public void listenWorkQueue1(String message) { System.out.println("listenWorkQueue1 message = " + message); } @RabbitListener(queues = "work.queue1") public void listenWorkQueue2(String message) { System.err.println("listenWorkQueue2... message = " + message); } }发送
@Test public void testWorkQueue() throws InterruptedException { String queueName = "work.queue1"; for (int i = 0; i < 50; i++) { rabbitTemplate.convertAndSend(queueName, i); Thread.sleep(20); } }结果
listenWorkQueue1 message = 0 listenWorkQueue2... message = 1 listenWorkQueue1 message = 2 listenWorkQueue2... message = 3 ...... listenWorkQueue1 message = 44 listenWorkQueue2... message = 45 listenWorkQueue1 message = 46 listenWorkQueue2... message = 47 listenWorkQueue1 message = 48 listenWorkQueue2... message = 49监听 加上延迟 模拟不同的性能
package com.itheima.consumer.Listeners; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * @Author：CharmingDaiDai * @Project：mq-demo * @Date：2024/4/27 下午2:59 */ @Slf4j @Component public class MqListener { @RabbitListener(queues = "learn.queue1") public void listen(String message) { System.out.println("message = " + message); } @RabbitListener(queues = "work.queue1") public void listenWorkQueue1(String message) throws InterruptedException { System.out.println("listenWorkQueue1 message = " + message); Thread.sleep(20); } @RabbitListener(queues = "work.queue1") public void listenWorkQueue2(String message) throws InterruptedException { System.err.println("listenWorkQueue2... message = " + message); Thread.sleep(200); } }结果
listenWorkQueue2... message = 0 listenWorkQueue1 message = 1 listenWorkQueue1 message = 3 listenWorkQueue1 message = 5 listenWorkQueue2... message = 2 listenWorkQueue1 message = 7 listenWorkQueue1 message = 9 listenWorkQueue1 message = 11 listenWorkQueue1 message = 13 listenWorkQueue2... message = 4 listenWorkQueue1 message = 15 listenWorkQueue1 message = 17 listenWorkQueue1 message = 19 listenWorkQueue2... message = 6 listenWorkQueue1 message = 21 listenWorkQueue1 message = 23 listenWorkQueue1 message = 25 listenWorkQueue2... message = 8 listenWorkQueue1 message = 27 listenWorkQueue1 message = 29 listenWorkQueue1 message = 31 listenWorkQueue2... message = 10 listenWorkQueue1 message = 33 listenWorkQueue1 message = 35 listenWorkQueue1 message = 37 listenWorkQueue1 message = 39 listenWorkQueue2... message = 12 listenWorkQueue1 message = 41 listenWorkQueue1 message = 43 listenWorkQueue1 message = 45 listenWorkQueue2... message = 14 listenWorkQueue1 message = 47 listenWorkQueue1 message = 49 listenWorkQueue2... message = 16 listenWorkQueue2... message = 18 listenWorkQueue2... message = 20 listenWorkQueue2... message = 22 listenWorkQueue2... message = 24 listenWorkQueue2... message = 26 listenWorkQueue2... message = 28 listenWorkQueue2... message = 30 listenWorkQueue2... message = 32 listenWorkQueue2... message = 34 listenWorkQueue2... message = 36 listenWorkQueue2... message = 38 listenWorkQueue2... message = 40 listenWorkQueue2... message = 42 listenWorkQueue2... message = 44 listenWorkQueue2... message = 46 listenWorkQueue2... message = 48
消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。导致1个消费者空闲，另一个消费者忙的不可开交。没有充分利用每一个消费者的能力，最终消息处理的耗时远远超过了1秒。这样显然是有问题的。
能者多劳
消费者添加配置
logging:
pattern:
dateformat: MM-dd HH:mm:ss:SSS
spring:
rabbitmq:
host: 10.61.2.143 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: learn # 虚拟主机
username: learn # 用户名
password: 1 # 密码
listener:
simple:
prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息结果
listenWorkQueue2... message = 0 listenWorkQueue1 message = 1 listenWorkQueue1 message = 2 listenWorkQueue1 message = 3 listenWorkQueue1 message = 4 listenWorkQueue1 message = 5 listenWorkQueue1 message = 6 listenWorkQueue1 message = 7 listenWorkQueue2... message = 8 listenWorkQueue1 message = 9 listenWorkQueue1 message = 10 listenWorkQueue1 message = 11 listenWorkQueue1 message = 12 listenWorkQueue1 message = 13 listenWorkQueue1 message = 14 listenWorkQueue1 message = 15 listenWorkQueue2... message = 16 listenWorkQueue1 message = 17 listenWorkQueue1 message = 18 listenWorkQueue1 message = 19 listenWorkQueue1 message = 20 listenWorkQueue1 message = 21 listenWorkQueue1 message = 22 listenWorkQueue1 message = 23 listenWorkQueue2... message = 24 listenWorkQueue1 message = 25 listenWorkQueue1 message = 26 listenWorkQueue1 message = 27 listenWorkQueue1 message = 28 listenWorkQueue1 message = 29 listenWorkQueue1 message = 30 listenWorkQueue1 message = 31 listenWorkQueue2... message = 32 listenWorkQueue1 message = 33 listenWorkQueue1 message = 34 listenWorkQueue1 message = 35 listenWorkQueue1 message = 36 listenWorkQueue1 message = 37 listenWorkQueue1 message = 38 listenWorkQueue1 message = 39 listenWorkQueue2... message = 40 listenWorkQueue1 message = 41 listenWorkQueue1 message = 42 listenWorkQueue1 message = 43 listenWorkQueue1 message = 44 listenWorkQueue1 message = 45 listenWorkQueue1 message = 46 listenWorkQueue1 message = 47 listenWorkQueue2... message = 48 listenWorkQueue1 message = 49
Fanout交换机
- Publisher：生产者，不再发送消息到队列中，而是发给交换机
- Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
- Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。
- Consumer：消费者，与以前一样，订阅队列，没有变化
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
交换机的类型有四种：
- Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
- Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列
- Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符
- Headers：头匹配，基于MQ的消息头匹配，用的较少。
Fanout交换机
- 1） 可以有多个队列
- 2） 每个队列都要绑定到Exchange（交换机）
- 3） 生产者发送的消息，只能发送到交换机
- 4） 交换机把消息发送给绑定过的所有队列
- 5） 订阅队列的消费者都能拿到消息
接收
package com.itheima.consumer.Listeners; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * @Author：CharmingDaiDai * @Project：mq-demo * @Date：2024/4/27 下午2:59 */ @Slf4j @Component public class MqListener { @RabbitListener(queues = "fanout.queue1") public void listenFanoutQueue1(String message) throws InterruptedException { System.out.println("listenWorkQueue1 message = " + message); } @RabbitListener(queues = "fanout.queue2") public void listenFanoutQueue2(String message) throws InterruptedException { System.err.println("listenWorkQueue2... message = " + message); } }发送
@Test public void testSendFanout() throws InterruptedException { String exchangeName = "learn.fanout"; String msg = "hello, every one!"; rabbitTemplate.convertAndSend(exchangeName, null, msg); }结果
listenWorkQueue1 message = hello, every one! listenWorkQueue2... message = hello, every one!
Direct交换机
在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在Direct模型下：
队列与交换机的绑定，不能是任意绑定了，而是要指定一个
RoutingKey（路由key）消息的发送方在 向 Exchange发送消息时，也必须指定消息的
RoutingKey。Exchange不再把消息交给每一个绑定的队列，而是根据消息的
Routing Key进行判断，只有队列的Routingkey与消息的Routing key完全一致，才会接收到消息添加两个队列
image-20240427181831290 创建交换机
image-20240427182127026 绑定 key、queue
image-20240427182115790 接收
package com.itheima.consumer.Listeners; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * @Author：CharmingDaiDai * @Project：mq-demo * @Date：2024/4/27 下午2:59 */ @Slf4j @Component public class MqListener { @RabbitListener(queues = "direct.queue1") public void listenDirectQueue1(String message) throws InterruptedException { System.out.println("listenQueue1 message = " + message); } @RabbitListener(queues = "direct.queue2") public void listenDirectQueue2(String message) throws InterruptedException { System.err.println("listenQueue2... message = " + message); } }发送
@Test public void testSendDirect() throws InterruptedException { String exchangeName = "learn.direct"; String msg = "hello, every one!"; rabbitTemplate.convertAndSend(exchangeName, "red", msg + "red"); rabbitTemplate.convertAndSend(exchangeName, "blue", msg + "blue"); rabbitTemplate.convertAndSend(exchangeName, "yellow", msg + "yellow"); }结果
listenQueue2... message = hello, every one!red listenQueue2... message = hello, every one!yellow listenQueue1 message = hello, every one!red listenQueue1 message = hello, every one!blue
Topic交换机
Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。
只不过Topic类型Exchange可以让队列在绑定BindingKey 的时候使用通配符！
BindingKey 一般都是有一个或多个单词组成，多个单词之间以.分割，例如： item.insert
通配符规则：
#：匹配0个或多个词*：匹配不多不少恰好1个词
举例：
item.#：能够匹配item.spu.insert或者item.spuitem.*：只能匹配item.spu
假如此时publisher发送的消息使用的RoutingKey共有四种：
china.news代表有中国的新闻消息；china.weather代表中国的天气消息；japan.news则代表日本新闻japan.weather代表日本的天气消息；
解释：
topic.queue1：绑定的是china.#，凡是以china.开头的routing key都会被匹配到，包括：china.newschina.weather
topic.queue2：绑定的是#.news，凡是以.news结尾的routing key都会被匹配。包括:china.newsjapan.news
添加两个队列
image-20240427192215506 添加交换机
image-20240427192249637 绑定
image-20240427192343896 接收
package com.itheima.consumer.Listeners; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * @Author：CharmingDaiDai * @Project：mq-demo * @Date：2024/4/27 下午2:59 */ @Slf4j @Component public class MqListener { @RabbitListener(queues = "topic.queue1") public void listenTopicQueue1(String message) throws InterruptedException { System.out.println("listenQueue1 message = " + message); } @RabbitListener(queues = "topic.queue2") public void listenTopicQueue2(String message) throws InterruptedException { System.err.println("listenQueue2... message = " + message); } }发送
@Test public void testSendTopic() throws InterruptedException { String exchangeName = "learn.topic"; String msg = "hello, every one!"; rabbitTemplate.convertAndSend(exchangeName, "china.cdd", msg + "china.cdd"); rabbitTemplate.convertAndSend(exchangeName, "cdd.news", msg + "cdd.news"); rabbitTemplate.convertAndSend(exchangeName, "china.news", msg + "china.news"); }结果
listenQueue1 message = hello, every one!china.cdd listenQueue1 message = hello, every one!china.news listenQueue2... message = hello, every one!cdd.news listenQueue2... message = hello, every one!china.news
声明队列和交换机的方式
基于Bean
一般是消费者端
SpringAMQP提供了几个类，用来声明队列、交换机及其绑定关系：
Queue:用于声明队列，可以用工厂类QueueBuilder构建
Exchange:用于声明交换机，可以用工厂类ExchangeBuilder构建
Binding:用于声明队列和交换机的绑定关系，可以用工厂类BindingBuilder构建
config.FanoutConfiguration
package com.itheima.consumer.config; import org.springframework.amqp.core.*; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * @Author：CharmingDaiDai * @Project：mq-demo * @Date：2024/4/27 下午7:42 */ @Configuration public class FanoutConfiguration { @Bean public FanoutExchange fanoutExchange() { // ExchangeBuilder.fanoutExchange("learn.fanout2").build(); return new FanoutExchange("learn.fanout2"); } @Bean public Queue fanoutQueue3() { // return QueueBuilder.durable("learn.fanout3").build(); // new Queue("learn.fanout3"); 默认是持久化的 return new Queue("learn.fanout3"); } @Bean public Binding fanoutBinding3(Queue fanoutQueue3, FanoutExchange fanoutExchange) { return BindingBuilder.bind(fanoutQueue3).to(fanoutExchange); } @Bean public Queue fanoutQueue4() { return new Queue("learn.fanout4"); } @Bean public Binding fanoutBinding4() { // fanoutQueue3() 方法上标注了@Bean注解，这会告诉Spring容器要执行该方法以创建相应的Bean对象。 // 如果已经存在了名为fanoutQueue3的Bean对象，则不会再执行fanoutQueue3()方法，而是直接使用已存在的Bean对象。 // 最后，将fanoutQueue3()方法返回的队列与fanoutExchange()方法返回的交换机进行绑定。 return BindingBuilder.bind(fanoutQueue4()).to(fanoutExchange()); } }
基于注解
liisteners.MqListener
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),
exchange = @Exchange(name = "learn.direct", type = ExchangeTypes.DIRECT),
key = {"red", "blue"}
))
public void listenDirectQueue1(String message) throws InterruptedException {
System.out.println("listenQueue1 message = " + message);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue2"),
exchange = @Exchange(name = "learn.direct", type = ExchangeTypes.DIRECT),
key = {"red", "yellow"}
))
public void listenDirectQueue2(String message) throws InterruptedException {
System.err.println("listenQueue2... message = " + message);
}
消息转换器
添加队列
image-20240428092401409 发送
@Test public void testSendObject(){ String queueName = "object.queue"; Map<String, Object> msg = new HashMap<>(2); msg.put("name", "jack"); msg.put("age", 18); rabbitTemplate.convertAndSend(queueName, msg); }结果
image-20240428093040794 原因
content_type:application/x-java-serialized-object
用的是java序列化，转换成字节
而在数据传输时，它会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。
只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：
- 数据体积过大
- 有安全漏洞
- 可读性差
配置JSON转换器
在publisher和consumer两个服务中都引入依赖：
<!--Jackson-->
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-xml</artifactId>
<version>2.17.0</version>
</dependency>注意，如果项目中引入了spring-boot-starter-web依赖，则无需再次引入Jackson依赖。
配置消息转换器，在publisher和consumer两个服务的启动类中添加一个Bean即可：
@Bean
public MessageConverter messageConverter(){
// 1.定义消息转换器
Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
// 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
jackson2JsonMessageConverter.setCreateMessageIds(true);
return jackson2JsonMessageConverter;
}消息转换器中添加的messageId可以便于我们将来做幂等性判断。
此时，我们到MQ控制台删除object.queue中的旧的消息。然后再次执行刚才的消息发送的代码，到MQ的控制台查看消息结构：
接收
@RabbitListener(queues = "object.queue") public void listenObjectQueue(Map<String, Object> message) throws InterruptedException { System.err.println("listenObjectQueue... message = " + message); }
listenObjectQueue... message = {name=jack, age=18}生产者可靠性
消息从生产者到消费者的每一步都可能导致消息丢失：
- 发送消息时丢失：
- 生产者发送消息时连接MQ失败
- 生产者发送消息到达MQ后未找到
Exchange - 生产者发送消息到达MQ的
Exchange后，未找到合适的Queue - 消息到达MQ后，处理消息的进程发生异常
- MQ导致消息丢失：
- 消息到达MQ，保存到队列后，尚未消费就突然宕机
- 消费者处理消息时：
- 消息接收后尚未处理突然宕机
- 消息接收后处理过程中抛出异常
综上，我们要解决消息丢失问题，保证MQ的可靠性，就必须从3个方面入手：
- 确保生产者一定把消息发送到MQ
- 确保MQ不会将消息弄丢
- 确保消费者一定要处理消息
生产者重试机制
生产者发送消息时，出现了网络故障，导致与MQ的连接中断
为了解决这个问题，SpringAMQP提供的消息发送时的重试机制。即：当RabbitTemplate与MQ连接超时后，多次重试。
修改publisher模块的application.yaml文件
logging:
pattern:
dateformat: MM-dd HH:mm:ss:SSS
spring:
rabbitmq:
host: 10.61.2.143 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: learn # 虚拟主机
username: learn # 用户名
password: 1 # 密码
connection-timeout: 1s # 设置MQ的连接超时时间
template:
retry:
enabled: true # 开启超时重试机制
initial-interval: 1000ms # 失败后的初始等待时间
multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multiplier
max-attempts: 3 # 最大重试次数然后重启rabbitMQ的docker
❗️注意：当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的。
如果对于业务性能有要求，建议禁用重试机制。如果一定要使用，请合理配置等待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码。
生产者确认机制
一般情况下，只要生产者与MQ之间的网路连接顺畅，基本不会出现发送消息丢失的情况，因此大多数情况下我们无需考虑这种问题。
不过，在少数情况下，也会出现消息发送到MQ之后丢失的现象，比如：
- MQ内部处理消息的进程发生了异常
- 生产者发送消息到达MQ后未找到
Exchange - 生产者发送消息到达MQ的
Exchange后，未找到合适的Queue，因此无法路由
针对上述情况，RabbitMQ提供了生产者消息确认机制，包括Publisher Confirm和Publisher Return两种。在开启确认机制的情况下，当生产者发送消息给MQ后，MQ会根据消息处理的情况返回不同的回执。
具体如图所示：
总结如下：
- 当消息投递到MQ，但是路由失败时，通过Publisher Return返回异常信息，同时返回ack的确认信息，代表投递成功
- 临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
- 持久消息投递到了MQ，并且入队完成持久化，返回ACK ，告知投递成功
- 其它情况都会返回NACK，告知投递失败
其中ack和nack属于Publisher Confirm机制，ack是投递成功；nack是投递失败。而return则属于Publisher Return机制。
默认两种机制都是关闭状态，需要通过配置文件来开启。
实现生产者确认
在publisher模块的application.yaml中添加配置：
logging:
pattern:
dateformat: MM-dd HH:mm:ss:SSS
spring:
rabbitmq:
host: 10.61.2.143 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: learn # 虚拟主机
username: learn # 用户名
password: 1 # 密码
connection-timeout: 1s # 设置MQ的连接超时时间
template:
retry:
enabled: true # 开启超时重试机制
initial-interval: 1000ms # 失败后的初始等待时间
multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multiplier
max-attempts: 3 # 最大重试次数
publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
publisher-returns: true # 开启publisher return机制这里publisher-confirm-type有三种模式可选：
none：关闭confirm机制simple：同步阻塞等待MQ的回执correlated：MQ异步回调返回回执
一般我们推荐使用correlated，回调机制。
- 定义ReturnCallBack
每个RabbitTemplate只能配置一个ReturnCallback，因此我们可以在配置类中统一设置。我们在publisher模块定义一个配置类：
config.MqConfirmConfig
package com.itheima.publisher.config; import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.core.ReturnedMessage; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Configuration; import javax.annotation.PostConstruct; /** * RabbitMQ消息确认配置类，用于设置消息发送失败时的回调函数。 * @author charmingdaidai */ @Slf4j @Configuration public class MqConfirmConfig { private final RabbitTemplate rabbitTemplate; /** * 构造函数注入RabbitTemplate。 * * @param rabbitTemplate RabbitTemplate对象 */ @Autowired public MqConfirmConfig(RabbitTemplate rabbitTemplate) { this.rabbitTemplate = rabbitTemplate; } /** * 初始化方法，在初始化时设置消息发送失败时的回调函数。 */ @PostConstruct public void init() { rabbitTemplate.setReturnsCallback(returned -> { log.error("触发return callback,"); log.debug("exchange: {}", returned.getExchange()); log.debug("routingKey: {}", returned.getRoutingKey()); log.debug("message: {}", returned.getMessage()); log.debug("replyCode: {}", returned.getReplyCode()); log.debug("replyText: {}", returned.getReplyText()); }); } }发送
@Test public void testConfirmCallback() throws InterruptedException { CorrelationData cd = new CorrelationData(UUID.randomUUID().toString()); cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() { @Override public void onFailure(Throwable ex) { log.error("消息回调失败", ex); } @Override public void onSuccess(CorrelationData.Confirm result) { log.debug("收到confirm callback回执"); if (result.isAck()){ log.debug("消息发送成功，收到ack"); } else { log.error("消息发送失败，收到nack，原因：{}", result.getReason()); } } }); rabbitTemplate.convertAndSend("learn.direct", "red", "hello", cd); Thread.sleep(2000); }04-28 14:09:43:363 INFO 40204 --- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [10.61.2.143:5672] 04-28 14:09:43:413 INFO 40204 --- [ main] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#1f346ad2:0/SimpleConnection@11f23203 [delegate=amqp://[email protected]:5672/learn, localPort= 4973] 04-28 14:09:43:439 DEBUG 40204 --- [0.61.2.143:5672] com.itheima.publisher.SpringAmqpTest : 收到confirm callback回执 04-28 14:09:43:439 DEBUG 40204 --- [0.61.2.143:5672] com.itheima.publisher.SpringAmqpTest : 消息发送成功，收到ack发送不存在的
rabbitTemplate.convertAndSend("learn.direct", "red2", "hello", cd);04-28 14:11:22:597 INFO 36720 --- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [10.61.2.143:5672] 04-28 14:11:22:645 INFO 36720 --- [ main] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#1f346ad2:0/SimpleConnection@11f23203 [delegate=amqp://[email protected]:5672/learn, localPort= 5051] 04-28 14:11:22:669 ERROR 36720 --- [nectionFactory1] c.i.publisher.config.MqConfirmConfig : 触发return callback, 04-28 14:11:22:669 DEBUG 36720 --- [0.61.2.143:5672] com.itheima.publisher.SpringAmqpTest : 收到confirm callback回执 04-28 14:11:22:669 DEBUG 36720 --- [0.61.2.143:5672] com.itheima.publisher.SpringAmqpTest : 消息发送成功，收到ack 04-28 14:11:22:669 DEBUG 36720 --- [nectionFactory1] c.i.publisher.config.MqConfirmConfig : exchange: learn.direct 04-28 14:11:22:670 DEBUG 36720 --- [nectionFactory1] c.i.publisher.config.MqConfirmConfig : routingKey: red2 04-28 14:11:22:670 DEBUG 36720 --- [nectionFactory1] c.i.publisher.config.MqConfirmConfig : message: (Body:'"hello"' MessageProperties [headers={spring_returned_message_correlation=e1dc69aa-0e84-4972-9358-c1a9bc6f05d7, __TypeId__=java.lang.String}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0]) 04-28 14:11:22:671 DEBUG 36720 --- [nectionFactory1] c.i.publisher.config.MqConfirmConfig : replyCode: 312 04-28 14:11:22:671 DEBUG 36720 --- [nectionFactory1] c.i.publisher.config.MqConfirmConfig : replyText: NO_ROUTE
❗️注意：
开启生产者确认比较消耗MQ性能，一般不建议开启。而且大家思考一下触发确认的几种情况：
- 路由失败：一般是因为RoutingKey错误导致，往往是编程导致
- 交换机名称错误：同样是编程错误导致
- MQ内部故障：这种需要处理，但概率往往较低。因此只有对消息可靠性要求非常高的业务才需要开启，而且仅仅需要开启ConfirmCallback处理nack就可以了。
MQ可靠性
数据持久化
为了提升性能，默认情况下MQ的数据都是在内存存储的临时数据，重启后就会消失。为了保证数据的可靠性，必须配置数据持久化，包括：
- 交换机持久化
- 在控制台的
Exchanges页面，添加交换机时可以配置交换机的Durability参数：
设置为Durable就是持久化模式，Transient就是临时模式。 - 队列持久化
- 在控制台的Queues页面，添加队列时，同样可以配置队列的
Durability参数： 
image-20240428142637876 - 消息持久化
- 在控制台发送消息的时候，可以添加很多参数，而消息的持久化是要配置一个
properties：
❗说明：在开启持久化机制以后，如果同时还开启了生产者确认，那么MQ会在消息持久化以后才发送ACK回执，进一步确保消息的可靠性。
不过出于性能考虑，为了减少IO次数，发送到MQ的消息并不是逐条持久化到数据库的，而是每隔一段时间批量持久化。一般间隔在100毫秒左右，这就会导致ACK有一定的延迟，因此建议生产者确认全部采用异步方式。
Lazy queue
在默认情况下，RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。但在某些特殊情况下，这会导致消息积压，比如：
- 消费者宕机或出现网络故障
- 消息发送量激增，超过了消费者处理速度
- 消费者处理业务发生阻塞
一旦出现消息堆积问题，RabbitMQ的内存占用就会越来越高，直到触发内存预警上限。此时RabbitMQ会将内存消息刷到磁盘上，这个行为称为PageOut。PageOut会耗费一段时间，并且会阻塞队列进程。因此在这个过程中RabbitMQ不会再处理新的消息，生产者的所有请求都会被阻塞。
为了解决这个问题，从RabbitMQ的3.6.0版本开始，就增加了Lazy Queues的模式，也就是惰性队列。惰性队列的特征如下：
- 接收到消息后直接存入磁盘而非内存
- 消费者要消费消息时才会从磁盘中读取并加载到内存（也就是懒加载）
- 支持数百万条的消息存储
而在3.12版本之后，LazyQueue已经成为所有队列的默认格式。因此官方推荐升级MQ为3.12版本或者所有队列都设置为LazyQueue模式。
代码配置方法
@Bean public Queue lazyQueue(){ return QueueBuilder .durable("lazy.queue") .lazy() // 开启Lazy模式 .build(); }注解配置方法
@RabbitListener(queuesToDeclare = @Queue( name = "lazy.queue", durable = "true", arguments = @Argument(name = "x-queue-mode", value = "lazy") )) public void listenLazyQueue(String msg){ log.info("接收到 lazy.queue的消息：{}", msg); }对于已经存在的队列，也可以配置为lazy模式，但是要通过设置policy实现。
可以基于命令行设置policy：rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues命令解读：
rabbitmqctl：RabbitMQ的命令行工具set_policy：添加一个策略Lazy：策略名称，可以自定义"^lazy-queue$"：用正则表达式匹配队列的名字'{"queue-mode":"lazy"}'：设置队列模式为lazy模式--apply-to queues：策略的作用对象，是所有的队列
消费者可靠性
当RabbitMQ向消费者投递消息以后，需要知道消费者的处理状态如何。因为消息投递给消费者并不代表就一定被正确消费了，可能出现的故障有很多，比如：
- 消息投递的过程中出现了网络故障
- 消费者接收到消息后突然宕机
- 消费者接收到消息后，因处理不当导致异常
- ...
一旦发生上述情况，消息也会丢失。因此，RabbitMQ必须知道消费者的处理状态，一旦消息处理失败才能重新投递消息。
但问题来了：RabbitMQ如何得知消费者的处理状态呢？
消费者确认机制
为了确认消费者是否成功处理消息，RabbitMQ提供了消费者确认机制（Consumer Acknowledgement）。即：当消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ自己消息处理状态。回执有三种可选值：
- ack：成功处理消息，RabbitMQ从队列中删除该消息
- nack：消息处理失败，RabbitMQ需要再次投递消息
- reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息
一般reject方式用的较少，除非是消息格式有问题，那就是开发问题了。因此大多数情况下我们需要将消息处理的代码通过try catch机制捕获，消息处理成功时返回ack，处理失败时返回nack.
由于消息回执的处理代码比较统一，因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式，有三种模式：
none：不处理。即消息投递给消费者后立刻ack，消息会立刻从MQ删除。非常不安全，不建议使用manual：手动模式。需要自己在业务代码中调用api，发送ack或reject，存在业务入侵，但更灵活auto：自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强，当业务正常执行时则自动返回ack. 当业务出现异常时，根据异常判断返回不同结果：- 如果是业务异常，会自动返回
nack； - 如果是消息处理或校验异常，自动返回
reject;
- 如果是业务异常，会自动返回
返回Reject的常见异常有：
Starting with version 1.3.2, the default ErrorHandler is now a ConditionalRejectingErrorHandler that rejects (and does not requeue) messages that fail with an irrecoverable error. Specifically, it rejects messages that fail with the following errors:
- o.s.amqp…MessageConversionException: Can be thrown when converting the incoming message payload using a MessageConverter.
- o.s.messaging…MessageConversionException: Can be thrown by the conversion service if additional conversion is required when mapping to a @RabbitListener method.
- o.s.messaging…MethodArgumentNotValidException: Can be thrown if validation (for example, @Valid) is used in the listener and the validation fails.
- o.s.messaging…MethodArgumentTypeMismatchException: Can be thrown if the inbound message was converted to a type that is not correct for the target method. For example, the parameter is declared as
Message<Foo>butMessage<Bar>is received.- java.lang.NoSuchMethodException: Added in version 1.6.3.
- java.lang.ClassCastException: Added in version 1.6.3.
通过下面的配置可以修改SpringAMQP的ACK处理方式：
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: none # 不做处理失败重试机制
当消费者出现异常后，消息会不断requeue（重入队）到队列，再重新发送给消费者。如果消费者再次执行依然出错，消息会再次requeue到队列，再次投递，直到消息处理成功为止。
极端情况就是消费者一直无法执行成功，那么消息requeue就会无限循环，导致mq的消息处理飙升，带来不必要的压力
为了应对上述情况Spring又提供了消费者失败重试机制：在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列。
修改consumer服务的application.yml文件，添加内容：
logging:
pattern:
dateformat: MM-dd HH:mm:ss:SSS
spring:
rabbitmq:
host: 10.61.2.143 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: learn # 虚拟主机
username: learn # 用户名
password: 1 # 密码
listener:
simple:
prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息
acknowledge-mode: none # 确认机制
retry:
enabled: true
initial-interval: 1000ms # 初识的失败等待时长为1秒
multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
max-attempts: 3 # 最大重试次数
stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false重启consumer服务，重复之前的测试。可以发现：
- 消费者在失败后消息没有重新回到MQ无限重新投递，而是在本地重试了3次
- 本地重试3次以后，抛出了
AmqpRejectAndDontRequeueException异常。查看RabbitMQ控制台，发现消息被删除了，说明最后SpringAMQP返回的是reject
结论：
- 开启本地重试时，消息处理过程中抛出异常，不会requeue到队列，而是在消费者本地重试
- 重试达到最大次数后，Spring会返回reject，消息会被丢弃
在之前的测试中，本地测试达到最大重试次数后，消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下，显然不太合适了。
因此Spring允许我们自定义重试次数耗尽后的消息处理策略，这个策略是由MessageRecovery接口来定义的，它有3个不同实现：
RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机
比较优雅的一种处理方案是RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工集中处理。
- config.ErrorConfiguration
package com.itheima.consumer.config;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @Author：CharmingDaiDai
* @Project：mq-demo
* @Date：2024/4/28 下午3:53
*/
@Configuration
@ConditionalOnProperty(prefix = "spring.rabbitmq.listener.simple.retry", name = "enabled", havingValue = "true")
public class ErrorConfiguration {
@Bean
public Queue errorQueue() {
return new Queue("error.queue");
}
@Bean
public DirectExchange errorExchange() {
return new DirectExchange("error.direct");
}
@Bean
public Binding errorBinding() {
return BindingBuilder.bind(errorQueue()).to(errorExchange()).with("error");
}
@Bean
public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}
}ℹ️@ConditionalOnProperty
@ConditionalOnProperty 是 Spring Boot 提供的一个条件注解，用于根据配置属性的值来决定是否加载某个 Bean 或者配置类。
在上面的示例中，@ConditionalOnProperty 注解标注在 ErrorConfiguration 类上，表示当满足一定条件时才会创建该类的实例（即将其作为一个 Bean 加载到 Spring 容器中）。
具体来说，该注解的属性含义如下：
prefix：配置属性的前缀，表示需要检查的配置属性的前缀。name：配置属性的名称，表示需要检查的配置属性的名称。havingValue：期望的配置属性值，当配置属性的值等于该值时，才会满足条件。默认情况下，属性值为字符串类型，如果需要比较其他类型的值，可以使用 Spring 的 SpEL 表达式。
在示例中，@ConditionalOnProperty(prefix = "spring.rabbitmq.listener.simple.retry", name = "enabled", havingValue = "true") 表示当 spring.rabbitmq.listener.simple.retry.enabled 配置属性的值为 true 时，才会创建 ErrorConfiguration 类的实例，即将其作为一个 Bean 加载到 Spring 容器中。
业务幂等性
何为幂等性？
幂等是一个数学概念，用函数表达式来描述是这样的：f(x) = f(f(x))，例如求绝对值函数。
在程序开发中，则是指同一个业务，执行一次或多次对业务状态的影响是一致的。例如：
- 根据id删除数据
- 查询数据
- 新增数据
在实际业务场景中，由于意外经常会出现业务被重复执行的情况，例如：
- 页面卡顿时频繁刷新导致表单重复提交
- 服务间调用的重试
- MQ消息的重复投递
我们在用户支付成功后会发送MQ消息到交易服务，修改订单状态为已支付，就可能出现消息重复投递的情况。如果消费者不做判断，很有可能导致消息被消费多次，出现业务故障。
举例：
- 假如用户刚刚支付完成，并且投递消息到交易服务，交易服务更改订单为已支付状态。
- 由于某种原因，例如网络故障导致生产者没有得到确认，隔了一段时间后重新投递给交易服务。
- 但是，在新投递的消息被消费之前，用户选择了退款，将订单状态改为了已退款状态。
- 退款完成后，新投递的消息才被消费，那么订单状态会被再次改为已支付。业务异常。
因此，我们必须想办法保证消息处理的幂等性。这里给出两种方案：
- 唯一消息ID
- 业务状态判断
唯一消息ID
生产 消费 都要配
给每个消息都设置一个唯一d,利用d区分是否是重复消息：
①每一条消息都生成一个唯一的id,与消息一起投递给消费者。
②消费者接收到消息后处理自己的业务，业务处理成功后将消息D保存到数据库
③如果下次又收到相同消息，去数据库查询判断是否存在，存在则为重复消息放弃处理。
@Bean
public MessageConverter messageConverter(){
// 1.定义消息转换器
Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
// 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
jjmc.setCreateMessageIds(true);
return jjmc;
}业务状态判断
业务判断就是基于业务本身的逻辑或状态来判断是否是重复的请求或消息，不同的业务场景判断的思路也不一样。
例如我们当前案例中，处理消息的业务逻辑是把订单状态从未支付修改为已支付。因此我们就可以在执行业务时判断订单状态是否是未支付，如果不是则证明订单已经被处理过，无需重复处理。
相比较而言，消息ID的方案需要改造原有的数据库，所以我更推荐使用业务判断的方案。
以支付修改订单的业务为例，我们需要修改OrderServiceImpl中的markOrderPaySuccess方法：
@Override
public void markOrderPaySuccess(Long orderId) {
// 1.查询订单
Order old = getById(orderId);
// 2.判断订单状态
if (old == null || old.getStatus() != 1) {
// 订单不存在或者订单状态不是1，放弃处理
return;
}
// 3.尝试更新订单
Order order = new Order();
order.setId(orderId);
order.setStatus(2);
order.setPayTime(LocalDateTime.now());
updateById(order);
}上述代码逻辑上符合了幂等判断的需求，但是由于判断和更新是两步动作，因此在极小概率下可能存在线程安全问题。
我们可以合并上述操作为这样：
// UPDATE `order` SET status = ? , pay_time = ? WHERE id = ? AND status = 1
@Override
public void markOrderPaySuccess(Long orderId) {
// UPDATE `order` SET status = ? , pay_time = ? WHERE id = ? AND status = 1
lambdaUpdate()
.set(Order::getStatus, 2)
.set(Order::getPayTime, LocalDateTime.now())
.eq(Order::getId, orderId)
.eq(Order::getStatus, 1)
.update();
}兜底方案
虽然我们利用各种机制尽可能增加了消息的可靠性，但也不好说能保证消息100%的可靠。万一真的MQ通知失败该怎么办呢？
有没有其它兜底方案，能够确保订单的支付状态一致呢？
其实思想很简单：既然MQ通知不一定发送到交易服务，那么交易服务就必须自己主动去查询支付状态。这样即便支付服务的MQ通知失败，我们依然能通过主动查询来保证订单状态的一致。
图中黄色线圈起来的部分就是MQ通知失败后的兜底处理方案，由交易服务自己主动去查询支付状态。
不过需要注意的是，交易服务并不知道用户会在什么时候支付，如果查询的时机不正确（比如查询的时候用户正在支付中），可能查询到的支付状态也不正确。
那么问题来了，我们到底该在什么时间主动查询支付状态呢？
这个时间是无法确定的，因此，通常我们采取的措施就是利用定时任务定期查询，例如每隔20秒就查询一次，并判断支付状态。如果发现订单已经支付，则立刻更新订单状态为已支付即可。
至此，消息可靠性的问题已经解决了。
综上，支付服务与交易服务之间的订单状态一致性是如何保证的？
- 首先，支付服务会正在用户支付成功以后利用MQ消息通知交易服务，完成订单状态同步。
- 其次，为了保证MQ消息的可靠性，我们采用了生产者确认机制、消费者确认、消费者失败重试等策略，确保消息投递的可靠性
- 最后，我们还在交易服务设置了定时任务，定期查询订单支付状态。这样即便MQ通知失败，还可以利用定时任务作为兜底方案，确保订单支付状态的最终一致性。
延迟消息
在电商的支付业务中，对于一些库存有限的商品，为了更好的用户体验，通常都会在用户下单时立刻扣减商品库存。例如电影院购票、高铁购票，下单后就会锁定座位资源，其他人无法重复购买。
但是这样就存在一个问题，假如用户下单后一直不付款，就会一直占有库存资源，导致其他客户无法正常交易，最终导致商户利益受损！
因此，电商中通常的做法就是：对于超过一定时间未支付的订单，应该立刻取消订单并释放占用的库存。
例如，订单支付超时时间为30分钟，则我们应该在用户下单后的第30分钟检查订单支付状态，如果发现未支付，应该立刻取消订单，释放库存。
但问题来了：如何才能准确的实现在下单后第30分钟去检查支付状态呢？
像这种在一段时间以后才执行的任务，我们称之为延迟任务，而要实现延迟任务，最简单的方案就是利用MQ的延迟消息了。
在RabbitMQ中实现延迟消息也有两种方案：
- 死信交换机+TTL
- 延迟消息插件
死信交换机
什么是死信？
当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：
- 消费者使用
basic.reject或basic.nack声明消费失败，并且消息的requeue参数设置为false - 消息是一个过期消息，超时无人消费
- 要投递的队列消息满了，无法投递
如果一个队列中的消息已经成为死信，并且这个队列通过dead-letter-exchange属性指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机就称为死信交换机（Dead Letter Exchange）。而此时加入有队列与死信交换机绑定，则最终死信就会被投递到这个队列中。
死信交换机有什么作用呢？
- 收集那些因处理失败而被拒绝的消息
- 收集那些因队列满了而被拒绝的消息
- 收集因TTL（有效期）到期的消息
延迟消息插件
Scheduling Messages with RabbitMQ | RabbitMQ
rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ (github.com)
查看RabbitMQ的插件目录对应的数据卷
(base) root@H3C-R4900-G5:/data1/wjm/projects# docker volume inspect mq-plugins [ { "CreatedAt": "2024-04-28T19:17:17+08:00", "Driver": "local", "Labels": null, "Mountpoint": "/data1/docker_home/volumes/mq-plugins/_data", "Name": "mq-plugins", "Options": null, "Scope": "local" } ]上传插件到该目录
(base) root@H3C-R4900-G5:/data1/wjm/projects# docker exec -it rabbitmq rabbitmq-plugins enable rabbitmq_delayed_message_exchange Enabling plugins on node rabbit@mq: rabbitmq_delayed_message_exchange The following plugins have been configured: rabbitmq_delayed_message_exchange rabbitmq_federation rabbitmq_management rabbitmq_management_agent rabbitmq_prometheus rabbitmq_web_dispatch Applying plugin configuration to rabbit@mq... The following plugins have been enabled: rabbitmq_delayed_message_exchange started 1 plugins.
声明方法
基于注解
@RabbitListener(bindings = @QueueBinding( value = @Queue(name = "delay.queue", durable = "true"), exchange = @Exchange(name = "delay.direct", delayed = "true"), key = "delay" )) public void listenDelayMessage(String msg){ log.info("接收到delay.queue的延迟消息：{}", msg); }基于Bean
package com.itheima.consumer.config; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.core.*; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Slf4j @Configuration public class DelayExchangeConfig { @Bean public DirectExchange delayExchange(){ return ExchangeBuilder .directExchange("delay.direct") // 指定交换机类型和名称 .delayed() // 设置delay的属性为true .durable(true) // 持久化 .build(); } @Bean public Queue delayedQueue(){ return new Queue("delay.queue"); } @Bean public Binding delayQueueBinding(){ return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay"); } }
发送消息
发送消息时，必须通过x-delay属性设定延迟时间：
@Test
void testPublisherDelayMessage() {
// 1.创建消息
String message = "hello, delayed message";
// 2.发送消息，利用消息后置处理器添加消息头
rabbitTemplate.convertAndSend("delay.direct", "delay", message, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
// 添加延迟消息属性
message.getMessageProperties().setDelay(5000);
return message;
}
});
}❗️注意：
延迟消息插件内部会维护一个本地数据库表，同时使用Elang Timers功能实现计时。如果消息的延迟时间设置较长，可能会导致堆积的延迟消息非常多，会带来较大的CPU开销，同时延迟消息的时间会存在误差。
因此，不建议设置延迟时间过长的延迟消息。
订单状态同步问题
假如订单超时支付时间为30分钟，理论上说我们应该在下单时发送一条延迟消息，延迟时间为30分钟。这样就可以在接收到消息时检验订单支付状态，关闭未支付订单。
但是大多数情况下用户支付都会在1分钟内完成，我们发送的消息却要在MQ中停留30分钟，额外消耗了MQ的资源。因此，我们最好多检测几次订单支付状态，而不是在最后第30分钟才检测。
例如：我们在用户下单后的第10秒、20秒、30秒、45秒、60秒、1分30秒、2分、...30分分别设置延迟消息，如果提前发现订单已经支付，则后续的检测取消即可。
这样就可以有效避免对MQ资源的浪费了。
定义常量
无论是消息发送还是接收都是在交易服务完成，因此我们在trade-service中定义一个常量类，用于记录交换机、队列、RoutingKey等常量：
内容如下：
package com.hmall.trade.constants;
public interface MqConstants {
String DELAY_EXCHANGE = "trade.delay.topic";
String DELAY_ORDER_QUEUE = "trade.order.delay.queue";
String DELAY_ORDER_ROUTING_KEY = "order.query";
}抽取共享mq配置
我们将mq的配置抽取到nacos中，方便各个微服务共享配置。
在nacos中定义一个名为shared-mq.xml的配置文件，内容如下：
spring:
rabbitmq:
host: 10.61.2.143 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: learn # 虚拟主机
username: learn # 用户名
password: 1 # 密码
listener:
simple:
prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息这里只添加一些基础配置，至于生产者确认，消费者确认配置则由微服务根据业务自己决定。
在trade-service模块添加共享配置：
4.5.3.改造下单业务
接下来，我们改造下单业务，在下单完成后，发送延迟消息，查询支付状态。
1）引入依赖
在trade-service模块的pom.xml中引入amqp的依赖：
<!--amqp-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>2）改造下单业务
修改trade-service模块的com.hmall.trade.service.impl.OrderServiceImpl类的createOrder方法，添加消息发送的代码：
4.5.4.编写查询支付状态接口
由于MQ消息处理时需要查询支付状态，因此我们要在pay-service模块定义一个这样的接口，并提供对应的FeignClient.
首先，在hm-api模块定义三个类：
说明：
- PayOrderDTO：支付单的数据传输实体
- PayClient：支付系统的Feign客户端
- PayClientFallback：支付系统的fallback逻辑
PayOrderDTO代码如下：
package com.hmall.api.dto;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.time.LocalDateTime;
/**
* <p>
* 支付订单
* </p>
*/
@Data
@ApiModel(description = "支付单数据传输实体")
public class PayOrderDTO {
@ApiModelProperty("id")
private Long id;
@ApiModelProperty("业务订单号")
private Long bizOrderNo;
@ApiModelProperty("支付单号")
private Long payOrderNo;
@ApiModelProperty("支付用户id")
private Long bizUserId;
@ApiModelProperty("支付渠道编码")
private String payChannelCode;
@ApiModelProperty("支付金额，单位分")
private Integer amount;
@ApiModelProperty("付类型，1：h5,2:小程序，3：公众号，4：扫码，5：余额支付")
private Integer payType;
@ApiModelProperty("付状态，0：待提交，1:待支付，2：支付超时或取消，3：支付成功")
private Integer status;
@ApiModelProperty("拓展字段，用于传递不同渠道单独处理的字段")
private String expandJson;
@ApiModelProperty("第三方返回业务码")
private String resultCode;
@ApiModelProperty("第三方返回提示信息")
private String resultMsg;
@ApiModelProperty("支付成功时间")
private LocalDateTime paySuccessTime;
@ApiModelProperty("支付超时时间")
private LocalDateTime payOverTime;
@ApiModelProperty("支付二维码链接")
private String qrCodeUrl;
@ApiModelProperty("创建时间")
private LocalDateTime createTime;
@ApiModelProperty("更新时间")
private LocalDateTime updateTime;
}PayClient代码如下：
package com.hmall.api.client;
import com.hmall.api.client.fallback.PayClientFallback;
import com.hmall.api.dto.PayOrderDTO;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
@FeignClient(value = "pay-service", fallbackFactory = PayClientFallback.class)
public interface PayClient {
/**
* 根据交易订单id查询支付单
* @param id 业务订单id
* @return 支付单信息
*/
@GetMapping("/pay-orders/biz/{id}")
PayOrderDTO queryPayOrderByBizOrderNo(@PathVariable("id") Long id);
}PayClientFallback代码如下：
package com.hmall.api.client.fallback;
import com.hmall.api.client.PayClient;
import com.hmall.api.dto.PayOrderDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.openfeign.FallbackFactory;
@Slf4j
public class PayClientFallback implements FallbackFactory<PayClient> {
@Override
public PayClient create(Throwable cause) {
return new PayClient() {
@Override
public PayOrderDTO queryPayOrderByBizOrderNo(Long id) {
return null;
}
};
}
}最后，在pay-service模块的PayController中实现该接口：
@ApiOperation("根据id查询支付单")
@GetMapping("/biz/{id}")
public PayOrderDTO queryPayOrderByBizOrderNo(@PathVariable("id") Long id){
PayOrder payOrder = payOrderService.lambdaQuery().eq(PayOrder::getBizOrderNo, id).one();
return BeanUtils.copyBean(payOrder, PayOrderDTO.class);
}4.5.5.消息监听
接下来，我们在trader-service编写一个监听器，监听延迟消息，查询订单支付状态：
代码如下：
package com.hmall.trade.listener;
import com.hmall.api.client.PayClient;
import com.hmall.api.dto.PayOrderDTO;
import com.hmall.common.domain.MultiDelayMessage;
import com.hmall.trade.constants.MqConstants;
import com.hmall.trade.domain.po.Order;
import com.hmall.trade.service.IOrderService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@RequiredArgsConstructor
public class OrderStatusListener {
private final IOrderService orderService;
private final PayClient payClient;
private final RabbitTemplate rabbitTemplate;
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = MqConstants.DELAY_ORDER_QUEUE, durable = "true"),
exchange = @Exchange(name = MqConstants.DELAY_EXCHANGE, type = ExchangeTypes.TOPIC),
key = MqConstants.DELAY_ORDER_ROUTING_KEY
))
public void listenOrderCheckDelayMessage(MultiDelayMessage<Long> msg) {
// 1.获取消息中的订单id
Long orderId = msg.getData();
// 2.查询订单，判断状态：1是未支付，大于1则是已支付或已关闭
Order order = orderService.getById(orderId);
if (order == null || order.getStatus() > 1) {
// 订单不存在或交易已经结束，放弃处理
return;
}
// 3.可能是未支付，查询支付服务
PayOrderDTO payOrder = payClient.queryPayOrderByBizOrderNo(orderId);
if (payOrder != null && payOrder.getStatus() == 3) {
// 支付成功，更新订单状态
orderService.markOrderPaySuccess(orderId);
return;
}
// 4.确定未支付，判断是否还有剩余延迟时间
if (msg.hasNextDelay()) {
// 4.1.有延迟时间，需要重发延迟消息，先获取延迟时间的int值
int delayVal = msg.removeNextDelay().intValue();
// 4.2.发送延迟消息
rabbitTemplate.convertAndSend(MqConstants.DELAY_EXCHANGE, MqConstants.DELAY_ORDER_ROUTING_KEY, msg,
message -> {
message.getMessageProperties().setDelay(delayVal);
return message;
});
return;
}
// 5.没有剩余延迟时间了，说明订单超时未支付，需要取消订单
orderService.cancelOrder(orderId);
}
}注意，这里要在OrderServiceImpl中实现cancelOrder方法