RabbitMQ的简介、安装、springBoot整合RabbitMQ、消息确认机制、延时队列-54

亦凉 2024-04-01 09:43 23阅读 0赞

1.目前我们已经完成了商品和搜索系统的开发。我们思考一下，是否存在问题？

*  商品的原始数据保存在数据库中，增删改查都在数据库中完成。
 *  搜索服务数据来源是索引库，如果数据库商品发生变化，索引库数据不能及时更新。

2.如果我们在后台修改了商品的价格，搜索页面依然是旧的价格，这样显然不对。该如何解决？这里有两种解决方案：

*  方案1：每当后台对商品做增删改操作，同时要修改索引库数据
 *  方案2：搜索服务对外提供操作接口，后台在商品增删改后，调用接口

**以上两种方式都有同一个严重问题：就是代码耦合，后台服务中需要嵌入搜索和商品页面服务，违背了微服务的`独立`原则。  
所以，我们会通过另外一种方式来解决这个问题：消息队列**

## 一：消息中间件简介 ##

### 1.作用 ###

**解耦：**  
在项目启动之初是很难预测未来会遇到什么困难的，消息中间件在处理过程中插入了一个隐含的，基于数据的接口层，两边都实现这个接口，这样就允许独立的修改或者扩展两边的处理过程，只要两边遵守相同的接口约束即可。  
**冗余（存储）：**  
在某些情况下处理数据的过程中会失败，消息中间件允许把数据持久化知道他们完全被处理  
**扩展性：**  
消息中间件解耦了应用的过程，所以提供消息入队和处理的效率是很容易的，只需要增加处理流程就可以了。  
**流量控制，削峰：**  
在访问量剧增的情况下，但是应用仍然需要发挥作用，但是这样的突发流量并不常见。而使用消息中间件采用队列的形式可以减少突发访问压力，不会因为突发的超时负荷要求而崩溃  
**可恢复性：**  
当系统一部分组件失效时，不会影响到整个系统。消息中间件降低了进程间的耦合性，当一个处理消息的进程挂掉后，加入消息中间件的消息仍然可以在系统恢复后重新处理  
**顺序保证：**  
在大多数场景下，处理数据的顺序也很重要，大部分消息中间件支持一定的顺序性  
**缓冲：**  
消息中间件通过一个缓冲层来帮助任务最高效率的执行  
**异步通信：**  
通过把把消息发送给消息中间件，消息中间件并不立即处理它，后续在慢慢处理。

### 2.特点 ###

#### 1）分布式 ####

消息中间件都是分布式的，因此才可以提供异步、解耦等功能。

#### 2）可靠性 ####

基于消息的通信是可靠的，消息不会丢失。大多数消息中间件都提供将消息持久化到磁盘的功能。

#### 3）异步 ####

通过消息中间件，可将远程同步调用拆解成为异步调用。对于不需要获取远程调用结果的应用场景来说，性能提升明 显。

#### 4）松耦合 ####

消息直接由中间件存储和分发。消息生产者只需关注如何将消息发送给消息中介服务器；消费者只需关注如何从中介服务器订阅。生产者和消费者之间是完全解耦的，不需要知道彼此的存在。

#### 5）事件驱动 ####

可以将复杂的应用系统重构成为事件驱动的系统

### 3.概述 ###

消息队列是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。

结合前面所说的问题：

*  商品服务对商品增删改以后，无需去操作索引库，只是发送一条消息，也不关心消息被谁接收。
 *  搜索服务服务接收消息，去处理索引库。

如果以后有其它系统也依赖商品服务的数据，同样监听消息即可，商品服务无需任何代码修改。

#### 1）大多应用中，可通过消息服务中间件来提升系统异步通信、扩展解耦能力 ####

#### 2）消息服务中两个重要概念： ####

**消息代理（message broker）和目的地（destination）**。当消息发送者发送消息以后，将由消息代理接管，消息代理保证消息传递到指定目的地。

#### 3）消息队列主要有两种形式的目的地 ####

队列（queue）：点对点消息通信（point-to-point）  
主题（topic）：发布（publish）/订阅（subscribe）消息通信

#### 4）点对点式 ####

*  消息发送者发送消息，消息代理将其放入一个队列中，消息接收者从队列中获 取消息内容，消息读取后被移出队列
 *  消息只有唯一的发送者和接受者，但并不是说只能有一个接收者

#### 5）发布订阅式 ####

发送者（发布者）发送消息到主题，多个接收者（订阅者）监听（订阅）这个 主题，那么就会在消息到达时同时收到消息

#### 6）JMS（Java Message Service）JAVA消息服务 ####

基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现

#### 7）AMQP（Advanced Message Queuing Protocol） ####

*  高级消息队列协议，也是一个消息代理的规范，兼容JMS
 *  RabbitMQ是AMQP的实现

#### 注：JMS和AMQP两者间的区别和联系： ####

*  JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
 *  JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
 *  JMS规定了两种消息模型；而AMQP的消息模型更加丰富  
    ![在这里插入图片描述][cdfda22e5d304de79cd93c2775f8dcfe.png]

#### 8）Spring支持 ####

*  spring-jms提供了对JMS的支持
 *  spring-rabbit提供了对AMQP的支持
 *  需要ConnectionFactory的实现来连接消息代理
 *  提供JmsTemplate、RabbitTemplate来发送消息
 *  @JmsListener（JMS）、@RabbitListener（AMQP）注解在方法上监听消息 代理发布的消息
 *  @EnableJms、@EnableRabbit开启支持

#### 9）Spring Boot自动配置 ####

*  JmsAutoConfiguration
 *  RabbitAutoConfiguration

#### 10）市面的MQ产品 ####

ActiveMQ、RabbitMQ、RocketMQ、Kafka

## 二：RabbitMQ简介和概述 ##

### 1.简介 ###

RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。  
优点：由于erlang语言的高并发特性，性能较好；吞吐量到万级，MQ功能比较完备，健壮、稳定、易用、跨平台、支持多种语言，支持AJAX文档齐全，开源提供的管理界面非常棒，用起来很好用，社区活跃度高；更新频率相当高。  
缺点：商业版需要收费，学习成本较高。

### 2.核心概念 ###

#### 1）Message——消息 ####

消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成， 这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可 能需要持久性存储）等。

#### 2）Publisher——生产者 ####

消息的生产者，也是一个向交换器发布消息的客户端应用程序。

#### 3）Exchange——交换器 ####

交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。  
Exchange有4种类型：direct(默认)，fanout, topic, 和headers，不同类型的Exchange转发消息的策略有所区别

#### 4）Queue——消息队列 ####

消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直 在队列里面，等待消费者连接到这个队列将其取走。

#### 5）Consumer——消费者 ####

消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

#### 6）Binding——绑定 ####

绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交 换器理解成一个由绑定构成的路由表。  
Exchange 和Queue的绑定可以是多对多的关系。

#### 7）Channel——信道 ####

信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP 命令都是通过信道 发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都 是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。

#### 8）Virtual Host——虚拟主机 ####

虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加 密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥 有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时 指定，RabbitMQ 默认的 vhost 是 / 。

#### 9）Broker——服务器实体 ####

表示消息队列服务器实体  
![在这里插入图片描述][b23a848d90954d0ca962861441c5dd0d.png]

## 三：docker容器安装RabbitMQ ##

### 1.下载 ###

官网下载地址：[http://www.rabbitmq.com/download.html][http_www.rabbitmq.com_download.html]  
官方文档：[https://www.rabbitmq.com/networking.html][https_www.rabbitmq.com_networking.html]

### 2.安装 ###

#### 2.1 docker容器下载RabbitMQ ####

docker pull rabbitmq:management

![在这里插入图片描述][45260909f4af40c4a806cfeff5b020d3.png]

#### 2.2 创建实例并启动 ####

docker run -d --name rabbitmq --publish 5671:5671 --publish 5672:5672 --publish 4369:4369 --publish 25672:25672 --publish 15671:15671 --publish 15672:15672 rabbitmq:management

注：  
4369 – erlang发现口  
5672 --client端通信口，AMQP端口  
15672 – 管理界面ui端口，web管理后台端口  
25672 – server间内部通信口，集群端口

#### 2.3 修改rabbitMQ为开机自启 ####

docker update rabbitmq --restart=always

#### 2.4 测试页面 ####

在web浏览器中输入地址：`http://你虚拟机ip:15672/`  
![!\[在这里插入图片描述\](https://img-blog.csdnimg.cn/6911fd2aa5384cab8aa1697fdc4ee545.png][https_img-blog.csdnimg.cn_6911fd2aa5384cab8aa1697fdc4ee545.png]  
输入默认账号: guest : guest  
注：如果服务器无法访问15672端口，记得在服务器中开通15672端口权限

#### 2.5 页面参数 ####

![在这里插入图片描述][ac460a6154c548a4a83f1cda11fb6bbe.png]  
overview：概览，各种统计信息，图表，也可以通过export definitions导出配置，转移到其他mq里面  
connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况  
channels：通道，建立连接后，会形成通道，消息的投递获取依赖通道。  
Exchanges：交换机，用来实现消息的路由  
Queues：队列，即消息队列，消息存放在队列中，等待消费，消费后被移除队列。  
admin：mq的管理和设置，添加用户信息，分配虚拟主机等

#### 2.6 管理界面-admin ####

#### 2.6.1. 添加用户 ####

如果不使用guest，我们也可以自己创建一个用户  
1、 超级管理员(administrator)  
可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。  
2、 监控者(monitoring)  
可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)  
3、 策略制定者(policymaker)  
可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。  
4、 普通管理者(management)  
仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。  
5、 其他  
无法登陆管理控制台，通常就是普通的生产者和消费者。

#### 2.6.2. 创建Virtual Hosts ####

虚拟主机：类似于mysql中的database。他们都是以“/”开头

## 四：RabbitMQ运行机制——Exchange 类型 ##

**AMQP 中消息的路由过程和 Java 开发者熟悉的 JMS 存在一些差别，AMQP 中增加了 Exchange 和Binding 的角色。生产者把消息发布到 Exchange 上，消息最终到达队列并被消费者接收，而 Binding 决定交 换器的消息应该发送到那个队列。**  
![在这里插入图片描述][7aec8308631b45129aab50a003f7ce0c.png]  
**Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键，headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了，所以直接 看另外三种类型：**

### 4.1 direct Exchange ###

![在这里插入图片描述][3d53234bbe5c402bb1dd473e5216f7db.png]  
消息中的路由键（routing key）如果和Binding 中的 binding key 一致， 交换器 就将消息发到对应的队列中。路由键与队 列名完全匹配，如果一个队列绑定到交换 机要求路由键为“dog”，则只转发 routing key 标记为“dog”的消息，不会转发“dog.puppy”，也不会转发“dog.guard” 等等。它是完全匹配、单播的模式。

### 4.2 fanout Exchange ###

![在这里插入图片描述][aa421c5d1d5a466ead4141a5a585440b.png]  
每个发到 fanout 类型交换器的消息都 会分到所有绑定的队列上去。fanout 交 换器不处理路由键，只是简单的将队列 绑定到交换器上，每个发送到交换器的 消息都会被转发到与该交换器绑定的所 有队列上。很像子网广播，每台子网内 的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。

### 4.3 topic Exchange ###

![在这里插入图片描述][da92d26802584d5e964081b6c28385fc.png]  
topic 交换器通过模式匹配分配消息的 路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。 它将路由键和绑定键的字符串切分成单 词，这些单词之间用点隔开。它同样也 会识别两个通配符：符号“\#”和符号“\*”。\#匹配0个或多个单词，\*匹配一 个单词。

### 4.4 direct直接交换机演示 ###

![在这里插入图片描述][080f6f69500b48f4aea2b4e3eb08aff9.png]  
1）首先创建出四个队列，如下图  
![在这里插入图片描述][dd93d4836aae4400ae45686eb07f7563.png]  
2）创建direct类型的交换机  
![在这里插入图片描述][96653b8533fb4214b4e48c7ecf4978b1.png]  
3）将exchange.direct交换机与四个队列进行绑定  
![在这里插入图片描述][445099f3c71346a3970d7ff7d9335634.png]  
4）在交换机里测试给sysg.news发送消息  
![在这里插入图片描述][ff1298fde17d4eb09afc14845562bf1a.png]  
5）此时消息就接受到了  
![在这里插入图片描述][64b0aba6b9a24b38887fbaf9ac03a389.png]  
注：  
nack message requeue true ： 接收消息，并且不在队列中删除  
automatic ack ：获取玩消息后，在队列中删除消息，保证消费消息

### 4.5 fanout交换机演示 ###

1）创建exchange.fanout交换机  
![在这里插入图片描述][5097e462ed0e408ba57962a64766d468.png]  
2）给交换机绑定如下四个队列  
![在这里插入图片描述][d131c775345a41e086d3765eb360d523.png]  
3）给sysg.emps队列发送消息  
![在这里插入图片描述][34971b7442864dc3b48ef095a6caeaeb.png]  
4）查看哪些队列收到了消息  
![在这里插入图片描述][5d28c464564949e49412024069649fa8.png]  
此时发现四个队列都接受到了消息，所以fanout每个队列都会接收到消息

### 4.6 topic交换机演示 ###

1）创建exchange.topic交换机  
![在这里插入图片描述][7e3ac1931b364486a1291e52174d5c1d.png]  
2）给交换机绑定如下四个队列  
![!\[在这里插入图片描述\](https://img-blog.csdnimg.cn/b287e2bb0c60476787520ce3c5d1a32f.g][https_img-blog.csdnimg.cn_b287e2bb0c60476787520ce3c5d1a32f.g]  
注：路由键的名称需要特别注意一下，模糊匹配  
3）给sysg.newss队列发送消息  
![在这里插入图片描述][62d664e55e264054a1fc674269dafa42.png]  
4）查看哪些队列收到了消息  
![在这里插入图片描述][5380cacb5f294d86b6d60840dc57ca46.png]  
前三个包含sysg，第四个包含news。所以四个队列都能接收到消息

## 五：SpringBoot整合RabbitMQ ##

### 5.1.引入pom依赖 ###

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

引入依赖后，RabbitAutoConfiguration就会自动生效。给容器中自动引入了RabbitTemplate和AmqpAdmin

### 5.2 在application.properties文件配置RabbitMQ的连接信息 ###

spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.port=5672
    spring.rabbitmq.virtual-host=/

### 5.3 使用@EnableRabbit注解，开启rabbitmq的相关功能 ###

@EnableRabbit
    @SpringBootApplication
    public class GulimailOrderApplication {
        
        public static void main(String[] args) {
        
            SpringApplication.run(GulimailOrderApplication.class, args);
        }
    }

## 六：测试rabbitmq的使用 ##

### 6.1 使用AmqpAdmin创建直接交换机 ###

@SpringBootTest
    @Slf4j
    class GulimailOrderApplicationTests {
        
        @Autowired
        private AmqpAdmin amqpAdmin;
        @Test
        public void createExchange() {
        
            /**
             * 全参构造器
             * DirectExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments)
             * name:交换机名称
             * durable：是否持久化
             * autoDelete：是否自动删除
             * arguments:参数
             */
            DirectExchange directExchange = new DirectExchange("hello-java-exchange",true,false);
            amqpAdmin.declareExchange(directExchange);
            log.info("Exchange[{}]创建成功","hello-java-exchange");
        }
    }

* DirectExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments)
         * name:交换机名称
         * durable：是否持久化
         * autoDelete：是否自动删除
         * arguments:参数

### 6.2 创建队列 ###

/**
         * 创建队列
         */
        @Test
        public void createQueue(){
        
            /**
             * 全参构造器
             * Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
             * name:队列名称
             * durable：是否持久化
             * exclusive：是否排他，队列一般不排他，谁能接受到谁用
             * autoDelete：是否自动删除
             * arguments:参数
             */
            Queue queue = new Queue("hello-java-queue",true,false,false);
            amqpAdmin.declareQueue(queue);
            log.info("Queue[{}]创建成功","hello-java-queue");
        }

* Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
         * name:队列名称
         * durable：是否持久化
         * exclusive：是否排他，队列一般不排他，谁能接受到谁用
         * autoDelete：是否自动删除
         * arguments:参数

### 6.3 将创建好的交换机和队列进行绑定 ###

**将exchange指定的交换机和destination目的地进行绑定，使用routingKey作为路由键**

/**
         * 创建队列和交换机绑定
         */
        @Test
        public void createBinding(){
        
            /**
             * 全参构造器
             * Binding(String destination, Binding.DestinationType destinationType, String exchange, String routingKey, Map<String, Object> arguments)
             * destination:目的地
             * destinationType：目的地类型
             * exchange：交换机
             * routingKey：路由键
             * arguments:参数
             * 将exchange指定的交换机和destination目的地进行绑定，使用routingKey作为路由键
             */
            Binding binding = new Binding("hello-java-queue",
                    Binding.DestinationType.QUEUE,
                    "hello-java-exchange",
                    "hello.java",
                    null);
            amqpAdmin.declareBinding(binding);
            log.info("binding[{}]绑定成功","hello-java-binding");
        }

* Binding(String destination, Binding.DestinationType destinationType, String exchange, String routingKey, Map<String, Object> arguments)
         * destination:目的地
         * destinationType：目的地类型
         * exchange：交换机
         * routingKey：路由键
         * arguments:参数

### 6.4 发送消息 ###

1）注入rabbitTemplate

@Autowired
    private RabbitTemplate rabbitTemplate;

2）新建rabbitConfig配置类——用来将对象转化为json，进行消息发送

3）发送消息代码编写

@Autowired
        private RabbitTemplate rabbitTemplate;
        @Test
        public void sendMessageTest(){
        
            OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
            reasonEntity.setId(1L);
            reasonEntity.setCreateTime(new Date());
            reasonEntity.setName("哈哈哈");
            //1.发送任意类型的消息，如果发出去的消息是对象，我们会使用序列化机制，将对象写出去，对象必须实现序列化接口
            String msg = "Hello World";
            //2.发送对象类型的消息转化为json
            rabbitTemplate.convertAndSend("hello-java-exchange","hello.java",reasonEntity);
            log.info("消息发送完成{}",reasonEntity);
        }

## 七：测试接收消息——@RabbitListener和@RabbitHandler ##

*  @RabbitListener——标注在类和方法上
 *  @RabbitHandler——标注在方法上

### 7.1 监听消息——@RabbitListener ###

监听消息:使用@RabbitListener，必须有@EnableRabbit

/**
         * queues：声明需要监听的所有队列
         */
        @RabbitListener(queues = {
        "hello-java-queue"})
        public void recieveMessage(Object message){
        
            System.out.println("接收到消息的内容："+message+"==>类型：" + message.getClass());
        }

### 7.2 传递的参数直接定义为多种类型 ###

参数可以定义为以下类型

* 1、Message message:原生消息的详细信息。头＋体
     * 2、T<发送消息的类型> OrderReturnReasonEntity content
     * 3、 Channel channel：传输数据的通道

/**
         * queues：声明需要监听的所有队列
         * 参数可以定义为以下类型
         * 1、Message message:原生消息的详细信息。头＋体
         * 2、T<发送消息的类型> OrderReturnReasonEntity content
         * 3、 Channel channel：传输数据的通道
         */
        @RabbitListener(queues = {
        "hello-java-queue"})
        public void recieveMessage(Message message, OrderReturnReasonEntity content , Channel channel){
        
            byte[] body = message.getBody();
            //消息头属性信息
            MessageProperties properties = message.getMessageProperties();
            System.out.println("接收到消息的内容："+message+"==>类型："+message.getClass());
        }

### 7.3 服务启动了多个，同一个消息，只有一个客户端可以接受到 ###

@Test
        public void sendMessageTest(){
        
            //1.发送任意类型的消息，如果发出去的消息是对象，我们会使用序列化机制，将对象写出去，对象必须实现序列化接口
            String msg = "Hello World";
            //2.发送对象类型的消息转化为json
            for (int i = 0; i < 10 ; i++) {
        
                OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
                reasonEntity.setId(1L);
                reasonEntity.setCreateTime(new Date());
                reasonEntity.setName("哈哈哈" + i);
                rabbitTemplate.convertAndSend("hello-java-exchange","hello.java",reasonEntity);
                log.info("消息发送完成{}",reasonEntity);
            }
        }

### 7.4 @RabbitHandler和@RabbitListener区别 ###

*  @RabbitListener 和 @RabbitHandler 搭配使用
 *  @RabbitListener 可以标注在类上面，需配合 @RabbitHandler 注解一起使用
 *  @RabbitListener 标注在类上面表示当有收到消息的时候，就交给 @RabbitHandler 的方法处理，具体使用哪个方法处理，根据 MessageConverter 转换后的参数类型

## 八：RabbitMQ消息确认机制 ##

### 8.1 如何避免消息丢失？ ###

1） 消息的丢失，在MQ角度考虑，一般有三种途径：

*  生产者确认发送到MQ服务器（生产者确认机制）
 *  MQ服务器不丢数据（消息持久化）
 *  消费者确认消费掉消息（消费者确认机制）

2） 生产者/消费者保证消息不丢失有两种实现方式：

*  开启事务模式
 *  消息确认模式

3） 开启事务会大幅降低消息发送及接收效率，使用的相对较少，因此我们生产环境一般都采取消息确认模式，我们只是讲解消息确认模式及消息持久化

*  生产者的ACK机制。有时，业务处理成功，消息也发了，但是我们并不知道消息是否成功到达了rabbitmq，例如：由于网络等原因导致业务成功而消息发送失败，此时可以使用rabbitmq的发送确认功能，要求rabbitmq显式告知我们消息是否已成功发送。
 *  消费者的ACK机制。可以防止消费者丢失消息。

**生产者确认机制有很严重的性能问题，如果每秒钟只有几百的消息量，可以使用。所以，我们主要讲了消费者的消息确认机制。**

### 8.2 生产者的确认机制——可靠抵达 ###

![在这里插入图片描述][24cef716f36145aeb503559ff0d05987.png]

**1）confirmCallback—— broker服务器接收到消息就会回调**

*  在创建 connectionFactory 的时候设置 PublisherConfirms(true) 选项，开启confirmcallback

spring.rabbitmq.publisher-confirms=true     #在创建 connectionFactory 的时候设置 PublisherConfirms(true) 选项，开启confirmcallback。

*  设置确认回调

@Configuration
    public class MyRabbitConfig {
        
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        /**
         * 定制rabbitTemplate
         * @post MyRabbitConfig对象创建完成后，执行这个方法
         */
        @PostConstruct
        public void initRabbitTemplate(){
        
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){
        
                /**
                 * 只要消息抵达broker服务器，就会ack = true
                 * @param correlationData 当前消息的唯一关联数据，这是消费的唯一id
                 * @param b 消息是否成功接收到
                 * @param s 失败的原因
                 */
                @Override
                public void confirm(CorrelationData correlationData, boolean b, String s) {
        
                }
            });
        }

*  消息只要被 broker 接收到就会执行 confirmCallback，如果是 cluster 模式，需要所有broker 接收到才会调用 confirmCallback。  
    被 broker 接收到只能表示 message 已经到达服务器，并不能保证消息一定会被投递 到目标 queue 里。所以需要用到接下来的 returnCallback

**2）returnCallback——消息正确抵达队列进行回调**

*  开启发送端消息抵达队列的确认

spring.rabbitmq.publisher-returns=true  #开启发送端消息抵达队列的确认
    spring.rabbitmq.template.mandatory=true  #只要抵达队列，以异步发送优先回调我们这个returnConfig

*  设置确认回调

**只要消息没有投递给指定队列，就会触发这个失败的回调。成功不会触发**

/**
         * 定制rabbitTemplate
         * @post MyRabbitConfig对象创建完成后，执行这个方法
         */
        @PostConstruct
        public void initRabbitTemplate(){
        
            //设置确认回调
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){
        
                /**
                 * 只要消息抵达broker服务器，就会ack = true
                 * @param correlationData 当前消息的唯一关联数据，这是消费的唯一id
                 * @param b 消息是否成功接收到
                 * @param s 失败的原因
                 */
                @Override
                public void confirm(CorrelationData correlationData, boolean b, String s) {
        
    
                }
            });
            //设置消息抵达队列的确认回调
            rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
        
                /**
                 * 只要消息没有投递给指定队列，就会触发这个失败的回调。成功不会触发
                 * @param message 投递失败消息的详细信息
                 * @param i 回复的状态
                 * @param s 回复的文本内容
                 * @param s1 当时消息发送给那个交换机
                 * @param s2 当时消息用哪个路由键
                 */
                @Override
                public void returnedMessage(Message message, int i, String s, String s1, String s2) {
        
                }
            });
        }

*  confrim 模式只能保证消息到达 broker，不能保证消息准确投递到目标 queue 里。在有 些业务场景下，我们需要保证消息一定要投递到目标 queue 里，此时就需要用到return 退回模式。这样如果未能投递到目标 queue 里将调用 returnCallback ，可以记录下详细到投递数 据，定期的巡检或者自动纠错都需要这些数据。

3）ack——消费者的ack机制

## 九：消费者的ack机制 ##

### 9.1消费者获取到消息，成功处理，可以回复Ack给Broker ###

*  basic.ack用于肯定确认；broker将移除此消息
 *  basic.nack用于否定确认；可以指定broker是否丢弃此消息，可以批量
 *  basic.reject用于否定确认；同上，但不能批量

### 9.2 默认自动ack，消息被消费者收到，就会从broker的queue中移除 ###

### 9.3 queue无消费者，消息依然会被存储，直到消费者消费 ###

### 9.4 消费者收到消息，默认会自动ack。但是如果无法确定此消息是否被处理完成， 或者成功处理。我们可以开启手动ack模式 ###

*  消息处理成功，ack()，接受下一个消息，此消息broker就会移除
 *  消息处理失败，nack()/reject()，重新发送给其他人进行处理，或者容错处理后ack
 *  消息一直没有调用ack/nack方法，broker认为此消息正在被处理，不会投递给别人，此时客户 端断开，消息不会被broker移除，会投递给别人

### 9.5 手动确认模式 ###

spring.rabbitmq.listener.direct.acknowledge-mode=manual

手动确认模式，只要不确认的消息都不算处理

### 9.6 消息丢失处理 ###

1）try-catch捕获异常，将因为网络等其他原因的获取出来，然后重新发送。  
2）每一个操作记录在日志做好记录，保存到数据表中。定期扫描然后重新发送。  
3）消息交给队列，持久化处理。  
4）做好消息确认机制，生产者和消费者两端确认。手动ack

### 9.7 消息重复处理 ###

1）调用的相应接口设置为幂等的  
2）使用防重表

### 9.8 消息积压处理 ###

消息积压会导致服务器性能下降。消费者能力不足或者发送端发送的能力太大  
1）上线更多的消费端  
2）上线处理的消息的消费者，将消息保存到数据库

## 十：延时队列 ##

![在这里插入图片描述][805c6b066a414396940dd17696c553b1.png]

### 10.1 创建队列、交换机和绑定关系 ###

/**
     * @bean 容器中的Binding、Queue、Exchange都会自动创建（rabbitmq没有的情况下）
     */
    @Configuration
    public class MyMQConfig {
        
        /**
         * 队列
         * @return
         */
        @Bean
        public Queue orderDelayQueue(){
        
            //order.delay.queue
            HashMap<String, Object> arguments = new HashMap<>();
            /**
             * x-dead-letter-exchange: order-event-exchange
             * x-dead-letter-routing-key: order.release.order
             * x-message-ttl: 60000
             */
            arguments.put("x-dead-letter-exchange","order-event-exchange");
            arguments.put("x-dead-letter-routing-key","order.release.order");
            arguments.put("x-message-ttl",60000);
            return new Queue("order.delay.queue",true,false,false,arguments);
    
        }
    
        /**
         * 队列
         * @return
         */
        @Bean
        public Queue orderReleaseOrderQueue(){
        
            //order.release.order.queue
            return new Queue("order.release.order.queue",true,false,false);
        }
    
        /**
         * 交换机
         * @return
         */
        @Bean
        public Exchange orderEventExchange(){
        
            //String name, boolean durable, boolean autoDelete, Map<String, Object> arguments
            return new TopicExchange("order-event-exchange",true,false);
        }
    
        /**
         * 绑定关系
         * @return
         */
        @Bean
        public Binding orderCreateOrderBinding(){
        
            //String destination, Binding.DestinationType destinationType, String exchange, String routingKey, Map<String, Object> arguments
            return new Binding("order.delay.queue",
                    Binding.DestinationType.QUEUE,
                    "order-event-exchange",
                    "order.create.order",
                    null);
        }
    
        /**
         * 绑定关系
         * @return
         */
        @Bean
        public Binding orderReleaseOrderBinding(){
        
            //String destination, Binding.DestinationType destinationType, String exchange, String routingKey, Map<String, Object> arguments
            return new Binding("order.release.order.queue",
                    Binding.DestinationType.QUEUE,
                    "order-event-exchange",
                    "order.release.order",
                    null);
    
        }
    
    
    
    
    
    }

### 10.2 监听队列测试 ###

1）编写controller，给MQ发送消息

@Autowired
        private RabbitTemplate rabbitTemplate;
    
        @ResponseBody
        @GetMapping(value = "/test/createOrder")
        public String createOrderTest() {
        
    
            //订单下单成功
            OrderEntity orderEntity = new OrderEntity();
            orderEntity.setOrderSn(UUID.randomUUID().toString());
            orderEntity.setModifyTime(new Date());
    
            //给MQ发送消息
            rabbitTemplate.convertAndSend("order-event-exchange","order.create.order",orderEntity);
    
            return "ok";
        }

2）监听队列

/**
         * 监听队列
         */
        @RabbitListener(queues = "order.release.order.queue")
        public void listener(OrderEntity entity , Channel channel , Message message) throws IOException {
        
            System.out.println("收到过期的订单信息：准备关闭订单" + entity.getOrderSn());
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }

## 十一：引入RabbitMQ ##

### 11.1 引入依赖 ###

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

### 11.2 配置主机地址 ###

## rabbitmq的服务器地址
    spring.rabbitmq.host=192.168.0.1
    # 虚拟主机
    spring.rabbitmq.virtual-host=/

### 11.3 @EnableRabbit ###

在主启动类添加@EnableRabbit注解，开启rabbitmq

### 11.4 将rabbitmq发送的消息转化为Json ###

### 11.5 添加交换机、队列和绑定关系 ###

@Configuration
    public class MyRabbitConfig {
        
    
        /**
         * 用来将对象转化为json
         * @return
         */
        @Bean
        public MessageConverter messageConverter(){
        
            return new Jackson2JsonMessageConverter();
        }
    
        /**
         * 创建stockEventExchange交换机
         * @return
         */
        @Bean
        public Exchange stockEventExchange(){
        
            //使用topic类型的交换机
            //String name, boolean durable, boolean autoDelete, Map<String, Object> arguments
            return new TopicExchange("stock-event-exchange",true,false);
        }
    
        @Bean
        public Queue stockReleaseStockQueue(){
        
    
            return new Queue("stock.release.stock.queue",true,false,false);
        }
    
        /**
         * 延时队列
         * @return
         */
        @Bean
        public Queue stockDelayQueue(){
        
            HashMap<String, Object> arguments = new HashMap<>();
            /**
             * x-dead-letter-exchange: order-event-exchange
             * x-dead-letter-routing-key: order.release.order
             * x-message-ttl: 60000
             */
            arguments.put("x-dead-letter-exchange","order-event-exchange");
            arguments.put("x-dead-letter-routing-key","order.release.order");
            arguments.put("x-message-ttl",120000);
            return new Queue("stock.delay.queue",true,false,false,arguments);
        }
    
        /**
         * 交换机与延迟队列绑定
         * @return
         */
        @Bean
        public Binding stockLockedBinding() {
        
            return new Binding("stock.delay.queue",
                    Binding.DestinationType.QUEUE,
                    "stock-event-exchange",
                    "stock.locked",
                    null);
        }
    
        /**
         * 交换机与普通队列绑定
         * @return
         */
        @Bean
        public Binding stockReleaseBinding() {
        
            //String destination, DestinationType destinationType, String exchange, String routingKey,
            // 			Map<String, Object> arguments
            return new Binding("stock.release.stock.queue",
                    Binding.DestinationType.QUEUE,
                    "stock-event-exchange",
                    "stock.release.#",
                    null);
        }

### 11.6 库存解锁的场景 ###

1）下单成功，订单过期未支付。用户手动取消订单  
2）下订单成功，库存锁定成功。但是接下来的业务处理发生异常，导致业务整体回滚。之前锁定的库存就要自动解锁。

[cdfda22e5d304de79cd93c2775f8dcfe.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/16d2d0244a394dceb49dda4123e1043d.png
[b23a848d90954d0ca962861441c5dd0d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/90c838cc99c24f459b886cc6bac3b128.png
[http_www.rabbitmq.com_download.html]: http://www.rabbitmq.com/download.html
[https_www.rabbitmq.com_networking.html]: https://www.rabbitmq.com/networking.html
[45260909f4af40c4a806cfeff5b020d3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/43b5affe578a4566b32329703a6895fc.png
[https_img-blog.csdnimg.cn_6911fd2aa5384cab8aa1697fdc4ee545.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/fdc3e3617f294e518863fe9f82d2209d.png
[ac460a6154c548a4a83f1cda11fb6bbe.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/d9b0d87e580d485e9c00734b24648b79.png
[7aec8308631b45129aab50a003f7ce0c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/9e690780394e4bcb88e8e4950867fa17.png
[3d53234bbe5c402bb1dd473e5216f7db.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/8293f519d43740b3bdb5d9ab67eb7eb8.png
[aa421c5d1d5a466ead4141a5a585440b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/eafa463560264547b599c2a09e419d58.png
[da92d26802584d5e964081b6c28385fc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/768b5648fdac43f68a6d1371e3f4f7f6.png
[080f6f69500b48f4aea2b4e3eb08aff9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/ab0b7b19e4514571b7a5a25b6e921d00.png
[dd93d4836aae4400ae45686eb07f7563.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/1567886e708e40afbb5cbda673bc8a14.png
[96653b8533fb4214b4e48c7ecf4978b1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/39a0a37de8d441e78bb8e2a6133b9e73.png
[445099f3c71346a3970d7ff7d9335634.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/332dc0b863e746d59f4471a827246d82.png
[ff1298fde17d4eb09afc14845562bf1a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/fd47382e1c244e4a9cef2b9ecaa9b5ca.png
[64b0aba6b9a24b38887fbaf9ac03a389.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/0e0c8df9e8ee4b0cbc01dee6e28a4d9b.png
[5097e462ed0e408ba57962a64766d468.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/01056ac39ff042a59c389e6914cdb117.png
[d131c775345a41e086d3765eb360d523.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/e9e1375ca8094fcdb1432512fc0db992.png
[34971b7442864dc3b48ef095a6caeaeb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/625df21b58e741dc9aed2ccc03ab5c6f.png
[5d28c464564949e49412024069649fa8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/82ff134052af46f1929a6cffc8b14e12.png
[7e3ac1931b364486a1291e52174d5c1d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/77f0cff3928c4f0ca3c66602a92506c5.png
[https_img-blog.csdnimg.cn_b287e2bb0c60476787520ce3c5d1a32f.g]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/970761d2d0254d9588d5e62d733d933a.png
[62d664e55e264054a1fc674269dafa42.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/223ce2840aa64843a9b3ef4895ec14e2.png
[5380cacb5f294d86b6d60840dc57ca46.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/6761a4c17d8541c8a86fe6c263479108.png
[24cef716f36145aeb503559ff0d05987.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/5ff68a3942c04921a3164b482f06f1f6.png
[805c6b066a414396940dd17696c553b1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/9f125d90d5314005b3108ea86c67c8ca.png