【RabbitMQ】利用手册

RabbitMQ

同步调用

长处：时效性强，期待到结果后才返回
缺点：扩展性差、性能下降（调用链越长耗时越久）、级联失败问题（一个调用点卡住，后面的链路都不能实行）
异步调用

异步调用通常是基于消息通知的方式，包含三个脚色：
消息发送者：投递消息的人，就是原来的调用者
消息吸收者：吸收和处置惩罚消息的人，就是原来的服务提供者
消息署理：管理、暂存、转发消息，你可以把它明确成微佩服务器

长处：
耦合度低，拓展性强
异步调用，无需期待，性能好
故障隔离，下游服务故障不影响上游业务
缓存消息，流量削峰填谷
缺点：
不能立即得到调用结果，时效性差
不确定下游业务实行是否乐成
业务安全依靠于Broker（消息署理）的可靠性
MQ技术选型

RabbitMQ安装

• 查找镜像：docker search rabbitmq
  
  
  
  
• 拉取镜像：docker pull rabbitmq:3.8.19，指定拉取版本为3.18.19，如果不指定则默认拉取latest
  
  
  
  
• 查看镜像：docker images
  
  
  
  
• 启动镜像：设置账号登录为admin，登录暗码为admin，不指定镜像版本，默认启动rabbitmq:latest
  
  1. docker run \
  2. -e RABBITMQ_DEFAULT_USER=admin \
  3. -e RABBITMQ_DEFAULT_PASS=admin \
  4. -v mq-plugins:/plugins \
  5. --name mq \
  6. --hostname localhost \
  7. -p 15672:15672 \
  8. -p 5672:5672 \
  9. -d \
  10. rabbitmq:3.8.19

  复制代码
  
  
  • 15672是RabbitMQ的背景管理端口
    
  • 5672是RabbitMQ的AQMP端口
    
  • /plugins是RabbitMQ容器存放插件的路径
    
  
  
  
  
• 查看容器：docker ps
  
  
  
  
• 进入RabbitMQ容器：docker exec -it 4df /bin/bash
  
  
  
  
• 开启RabbitMQ背景访问：rabbitmq-plugins enable rabbitmq_management
  
  
  
  
• 退出容器bash：exit
  
  
  
  
• 网页访问RabbitMQ背景：访问http://localhost:15672，账号admin，暗码admin
  
  
  
  

常见问题：

• 背景管理系统的可视化界面中出现：All stable feature flags must be enabled after completing an upgrade
  **办理方案：**点击Admin -> Feature Flags，确保所有稳定的特性标志都是启用状态。如果有任何标志未启用，请将其启用。
  
  
  
  
• 背景管理系统的可视化界面中出现：Stats in management UI are disabled on this node
  
  
  
  **办理方案：**进入RabbitMQ容器，运行命令：echo management_agent.disable_metrics_collector=false>/etc/rabbitmq/conf.d/management_agent.disable_metrics_collector.conf，退出RabbitMQ容器，然后运行docker restart 容器id重启RabbitMQ容器。
  
  
  
  
• 背景管理系统的可视化界面中 Overview 不显示图形的问题
  办理方案：同《2. 背景管理系统的可视化界面中出现：Stats in management UI are disabled on this node》
  

RabbitMQ介绍

• publisher：消息发送者
  
• comsumer：消息消费者
  
• queue：队列-存储消息
  
• exchange：互换机-吸收发送者发送的消息，并将消息路由到与其绑定的队列
  
• virtual-host：虚拟主机-将数据隔离（多个项目利用同一个RabbitMQ时，可以为每个项目创建一个virtual-host，将不同项目之间的exchange和queue隔离）
  

Work Queues(任务模子)

任务模子简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。同一个消息只会被一个消费者处置惩罚。多个消费者绑定到一个队列，可以加快消息处置惩罚速度。

默认情况下，RabbitMQ的会将消息依次轮询投递给绑定在队列上的每一个消费者。但这并没有考虑到消费者是否已经处置惩罚完消息，可能出现消息堆积。因此我们需要修改消费者的application.yml，设置preFetch值为1，确保同一时候最多投递给消费者1条消息，消费者处置惩罚完后再投递下一条消息。

Fanout互换机

Fanout Exchange 会将吸收到的消息路由到每一个跟其绑定的queue，所以也叫广播模式。 Fanout互换机会将收到的消息复制成n份，然后将消息发送到n个与其绑定的队列中。

应用场景：用户付出乐成后，生意业务服务更新订单状态，短佩服务通知用户，积分服务为用户增长积分。
实现：生意业务服务的queue、短佩服务的queue、积分服务的queue都绑定到Fanout互换机，用户付出乐成后，付出服务将消息发送到Fanout互换机，这样生意业务服务、短佩服务、积分服务九都能收到这条消息了。
案例演示：
实现思路：

• 在RabbitMQ控制台中，声明队列fanout.queue1和fanout.queue2
  
• 在RabbitMQ控制台中，声明互换机hmall.fanout，将两个队列与其绑定
  
• 在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2
  
• 在publisher中编写测试方法，向hmall.fanout发送消息
  

代码实现：

发送者：

1. @SpringBootTest
2. class SpringAmqpTest {
3.     @Autowired
4.     private RabbitTemplate rabbitTemplate;
5.     @Test
6.     public void testSimpleQueue(){
7.         String exchangeName="hmall.fanout";
8.         String message="hello everyone";                                                    rabbitTemplate.convertAndSend(exchangeName,null,message);
9.     }
10. }

复制代码

消费者：

1. @Component
2. @Slf4j
3. public class SpringRabbitListener {
4.     @RabbitListener(queues = "fanout.queue1")
5.     public void listenerWorkQueue1(String message){
6.         log.info("消费者1接收到消息：{},---{}",message, LocalTime.now());
7.     }
8.     @RabbitListener(queues = "fanout.queue2")
9.     public void listenerWorkQueue2(String message){
10.         log.info("消费者2接收到消息：{},---{}",message,LocalTime.now());
11.     }
12. }

复制代码

消费者输出：

Direct互换机

Direct Exchange 会将吸收到的消息根据规则路由到指定的Queue，因此称为定向路由。

• 每一个Queue都与Exchange设置一个Bindingkey（可以为每一个Queue指定相同的Bindingkey，实现和Fanout互换机相同的功能）。
  
• 发布者发送消息时，指定消息的RoutingKey
  
• Exchange将消息路由到BindingKey与消息RoutingKey一致的队列
  

应用场景：用户取消后，只需要给生意业务服务发送消息，通知生意业务服务更新订单状态，而不需要给短佩服务和积分服务发送消息。
案例演示：
实现思路：

• 在RabbitMQ控制台中，声明队列direct.queue1和direct.queue2
  
• 在RabbitMQ控制台中，声明互换机hmall. direct，将两个队列与其绑定，routeKey 为blue时路由到direct.queue1，为yellow时路由到direct.queue2，为red时路由到direct.queue1和direct.queue2
  
• 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
  
• 在publisher中编写测试方法，利用不同的RoutingKey向hmall. direct发送消息
  

代码实现：

消费者

1. @Component
2. @Slf4j
3. public class SpringRabbitListener {
4.     @RabbitListener(queues = "direct.queue1")
5.     public void listenerWorkQueue1(String message){
6.         log.info("消费者1接收到消息：{},---{}",message, LocalTime.now());
7.     }
8.     @RabbitListener(queues = "direct.queue2")
9.     public void listenerWorkQueue2(String message){
10.         log.info("消费者2接收到消息：{},---{}",message,LocalTime.now());
11.     }
12. }

复制代码

发送者

1. @Autowired
2. private RabbitTemplate rabbitTemplate;
3. @Test
4. public void testSimpleQueue(){
5.   //交换机名称
6.   String exchangeName="hmall.direct";
7.   //消息
8.   String message_blue="hello blue";
9.   String message_yellow="hello yellow";
10.   String message_red="hello red";
11.   //发送消息
12. rabbitTemplate.convertAndSend(exchangeName,"blue",message_blue);
13. rabbitTemplate.convertAndSend(exchangeName,"yellow",message_yellow);
14. rabbitTemplate.convertAndSend(exchangeName,"red",message_red);
15. }

复制代码

消费者输出：

Topic互换机

TopicExchange也是基于routingkey做消息路由，但是routingkey通常是多个单词的组合，而且以.分割。
Queue与Exchange指定routingkey时可以利用通配符：

• #：代指0个或多个单词
  
• *：代指一个单词
  

案例演示：
实现思路：

• 在RabbitMQ控制台中，声明队列topic.queue1和topic.queue2
  
• 在RabbitMQ控制台中，声明互换机hmall. topic，将两个队列与其绑定
  
• 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
  
• 在publisher中编写测试方法，利用不同的RoutingKey向hmall. topic发送消息
  

代码实现：

消费者：

1. @Component
2. @Slf4j
3. public class SpringRabbitListener {
4.     @RabbitListener(queues = "topic.queue1")
5.     public void listenerWorkQueue1(String message){
6.         log.info("消费者1接收到消息：{},---{}",message, LocalTime.now());
7.     }
8.     @RabbitListener(queues = "topic.queue2")
9.     public void listenerWorkQueue2(String message){
10.         log.info("消费者2接收到消息：{},---{}",message,LocalTime.now());
11.     }
12. }

复制代码

发送者1：

1.     @Autowired
2.     private RabbitTemplate rabbitTemplate;
3.     @Test
4.     public void testSimpleQueue(){
5.         //交换机名称
6.         String exchangeName="hmall.topic";
7.         //消息
8.         String message="中国新闻";
9.         //发送消息
10.         rabbitTemplate.convertAndSend(exchangeName,"china.news",message);
11.     }

复制代码

消费者输出：

发送者2：

1.     @Autowired
2.     private RabbitTemplate rabbitTemplate;
3.     @Test
4.     public void testSimpleQueue(){
5.         //交换机名称
6.         String exchangeName="hmall.topic";
7.         //消息
8.         String message="中国天气";
9.         //发送消息
10.         rabbitTemplate.convertAndSend(exchangeName,"china.weather",message);
11.     }

复制代码

消费者输出：

AMQP

Advanced Message Queuing Protocol，是用于在应用步伐之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。
Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和吸收消息。包含两部分，其中spring-amqp是基础抽象，spring-rabbit是底层的默认实现。
AmqpTemplate和RabbitTemplate

AmqpTemplate 是一个接口，定义了基本的 AMQP 操纵，如发送消息、吸收消息、转换消息等。它提供了与 AMQP（包括 RabbitMQ）通讯的基本功能的抽象。
RabbitTemplate 是 AmqpTemplate 的默认实现类，专门用于与 RabbitMQ 举行交互。它实现了 AmqpTemplate 接口，并提供了更多与 RabbitMQ 交互的具体功能和配置选项。
RabbitTemplate 比 AmqpTemplate 更加丰富，提供了一些额外的高级特性和配置选项，如事件支持、消息确认机制、消息转换器等。这些功能可以更好地满意与 RabbitMQ 高级交互需求。
综上所述，AmqpTemplate 是一个通用的 AMQP 操纵接口，而 RabbitTemplate 是对其的具体实现，提供了更多与 RabbitMQ 交互的功能和默认配置，使得在 Spring 应用中利用 RabbitMQ 变得更加简单和方便。
RabbitMQ利用

背景可视化界面操纵

• 创建用户
  
  
  
  
• 创建虚拟主机
  
  
  
  
• 为用户添加可访问的虚拟主机
  
  
  
  
  
  
  注意：当前登录用户默认有权限访问其创建的所有虚拟主机。
  
• 创建队列
  
  
  
  
  
  
  • Durability：
    Durable：长期化队列，Rabbit服务器重启后，这个队列还会存在
    Transient：暂时队列，Rabbit服务器重启后，这个队列将会被删除
    
  
  
• 查看队列的消费者
  
  
  
  
• 向队列中发布消息
  
  
  
  
• 获取队列中消息
  队列中可以存储消息。当队列中的消息未被消费时，消息将存储在队列中，此时可以查看队列中的消息。
  
  
  
  
  
  
  • Act Mode：
    Nack message requeue true：获取消息，但是不做ack应答确认，消息重新入队
    Ack message requeue false：获取消息，应答确认，消息不重新入队，将会从队列中删除
    reject requeue true：拒绝获取消息，消息重新入队
    reject requeue false：拒绝获取消息，消息不重新入队，将会被删除
    
  • Encoding：可以选择将消息举行base64编码
    
  • Messages：从队列中获取的消息数量
    
  
  
• 清算消息
  
  
  
  

代码操纵

• 引入依靠
  
  1. <!-- AMQP依赖，包含RabbitMQ-->
  2. <dependency>
  3.   <groupId>org.springframework.boot</groupId>
  4.   <artifactId>spring-boot-starter-amqp</artifactId>
  5. </dependency>

  复制代码
• application.yaml中配置RabbitMQ
  
  1. spring:
  2.   rabbitmq:
  3.     host: 192.168.1.2 # RabbitMQ地址
  4.     port: 5672 # 端口
  5.     virtual-host: /hmall # 虚拟主机
  6.     username: jack # 用户名
  7.     password: jack # 密码

  复制代码

创建队列和互换机

SpringAMQP提供了几个类，用来声明队列、互换机及其绑定关系：

• Queue：用于声明队列，可以用工厂类QueueBuilder构建
  
• Exchange：用于声明互换机，可以用工厂类ExchangeBuilder构建
  
• Binding：用于声明队列和互换机的绑定关系，可以用工厂类BindingBuilder构建
  

如果已经存在互换机、队列、绑定关系，运行代码时则不会举行创建，而且也不会报错。
通常发送者只需要关心消息发送，消费者关心队列、互换机、以及绑定关系，所以创建操纵一样平常写在消费者中。
Sping提供了基于java bean和基于@RabbitListener注解两种方式创建。

• 基于bean代码演示：
  

1. package com.itheima.consumer.config;
3. import org.springframework.amqp.core.*;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.context.annotation.Configuration;
7. @Configuration
8. public class FanoutConfiguration {
9.   //声明交换机
10.     @Bean
11.     public FanoutExchange fanoutExchange(){
12. //        方式1
13. //        return new FanoutExchange("hmall.fanout");
14. //        方式2
15.         return ExchangeBuilder.fanoutExchange("hmall.fanout").build();
16.     }
17.   //声明队列
18.     @Bean
19.     public Queue fanoutQueue1(){
20. //        方式1
21. //        return new Queue("fanout.queue1",true);
22. //        方式2
23.         return QueueBuilder.durable("fanout.queue1").build();
24.     }
25.   //将队列和交换机绑定
26.     @Bean
27.     public Binding fanoutQueue1Binding(Queue fanoutQueue1,FanoutExchange fanoutExchange){//spring会去找beanName='fanoutQueue1'的bean作为参数传进来
28.         return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
29.     }
30.     @Bean
31.     public Queue fanoutQueue2(){
32.         return QueueBuilder.durable("fanout.queue2").build();
33.     }
34.     @Bean
35.     public Binding fanoutQueue2Binding(Queue fanoutQueue2,FanoutExchange fanoutExchange){//spring会去找beanName='fanoutQueue2'的bean作为参数传进来
36.         return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
37.     }
38. }

复制代码

• 基于@RabbitListener注解代码演示：
  

1. @Component
2. @Slf4j
3. public class SpringRabbitListener {
4.     @RabbitListener(bindings = @QueueBinding( //将交换机和队列绑定
5.             value = @Queue(name="direct.queue1",durable = "true"), //如果没有队列direct.queue1则创建队列,并监听队列direct.queue1
6.             exchange = @Exchange(name = "hmall.direct",type = ExchangeTypes.DIRECT), //如果没有交换机hmall.direct则创建交换机
7.             key = {"blue","red"} //routingKey
8.             ))
9.     public void listenerWorkQueue1(String message){
10.         log.info("消费者1接收到消息：{},---{}",message, LocalTime.now());
11.     }
13.     @RabbitListener(bindings = @QueueBinding(
14.             value = @Queue(name="direct.queue2",durable = "true"),
15.             exchange = @Exchange(name = "hmall.direct",type = ExchangeTypes.DIRECT),
16.             key = {"yellow","red"}
17.     ))
18.     public void listenerWorkQueue2(String message){
19.         log.info("消费者2接收到消息：{},---{}",message,LocalTime.now());
20.     }
21. }

复制代码

发送消息

• 直接发送给队列
  方法：public void convertAndSend(String routingKey, final Object object)，直接发给队列时，routingKey相称于队列名。
  
  1. @Autowired
  2. private RabbitTemplate rabbitTemplate；
  3. @Test
  4. public void testSimp leQueue() {
  5.   //队列名称
  6.   String queueName = "simple. queue";
  7.   //消息
  8.   String message = "hello, spring amqp!";
  9.   //发送消息
  10.   rabbitTemplate.convertAndSend(queueName, message)；
  11. }

  复制代码
  注意：队列不显示绑定互换机时，默认还是会绑定到defalut exchange上
  
  
  
  
• 发送给Fanout Exchange
  方法：public void convertAndSend(String exchange, String routingKey, final Object object)，利用Fanout Exchange时，routingKey相称于队列名，发送给Fanout Exchange时，routingKey传null或""
  
  1. @Autowired
  2. private RabbitTemplate rabbitTemplate;
  3. @Test
  4. public void testSimpleQueue(){
  5.   //交换机名称
  6.   String exchangeName="hmall.fanout";
  7.   //消息
  8.   String message="hello everyone";
  9.   //发送消息
  10. rabbitTemplate.convertAndSend(exchangeName,null,message);
  11. }

  复制代码
• 发送给direct互换机
  方法：public void convertAndSend(String exchange, String routingKey, final Object object)，routingKey就是互换机和队列绑定时的routingKey
  
  1. @Autowired
  2. private RabbitTemplate rabbitTemplate;
  3. @Test
  4. public void testSimpleQueue(){
  5.   //交换机名称
  6.   String exchangeName="hmall.direct";
  7.   //消息
  8.   String message_blue="hello blue";
  9.   String message_yellow="hello yellow";
  10.   String message_red="hello red";
  11.   //发送消息
  12. rabbitTemplate.convertAndSend(exchangeName,"blue",message_blue);
  13. rabbitTemplate.convertAndSend(exchangeName,"yellow",message_yellow);
  14. rabbitTemplate.convertAndSend(exchangeName,"red",message_red);
  15. }

  复制代码
• 发送给topic互换机
  方法：方法：public void convertAndSend(String exchange, String routingKey, final Object object)
  
  1. @Autowired
  2. private RabbitTemplate rabbitTemplate;
  3. @Test
  4. public void testSimpleQueue(){
  5.   //交换机名称
  6.   String exchangeName="hmall.topic";
  7.   //消息
  8.   String message="中国新闻";
  9.   //发送消息
  10.   rabbitTemplate.convertAndSend(exchangeName,"china.news",message);
  11. }

  复制代码

吸收消息

1. @Component
2. @Slf4j
3. public class SpringRabbitListener {
4.     @RabbitListener(queues = "队列名")
5.     public void listenerSimpleQueue(String message){
6.         log.info("消费者收到消息：{}",message);
7.     }
8. }

复制代码

配置消息转换器

convertAndSend方法会先将消息举行序列化，然后再发送。
Spring的对消息对象的处置惩罚是由org.springframework.amap.support.converter.Messageconverter来处置惩罚的。而
默认实现是SimpleMessageConverter，如果消息实现了Serializable接口，则会利用serialize方法举行序列化，而serialize方法是基于JDK的Objectoutputstream完成序列化的。存在下列问题：

• JDK的序列化有安全风险
  
• JDK序列化的消息太大
  
• JDK序列化的消息可读性差
  

建议采用JSON序列化取代默认的JDK序列化，要做两件事情：

• 在publisher和consumer中都要引入jackson依靠，发送者和消费者要利用相同的消息转换器:
  
  1. <dependency>
  2.   <groupId>com.fasterxml.jackson.core</groupId>
  3.   <artifactId>jackson-databind</artifactId>
  4. </dependency>

  复制代码
• 在publisher和consumer中都要配置MessageConverter：
  
  1. @Bean
  2. public MessageConverter messageConverter() {
  3.   return new Jackson2JsonMessageConverter();
  4. }

  复制代码

测试：

• 利用默认的消息转换器
  发送者：
  
  1. package com.itheima.publisher;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import org.junit.jupiter.api.Test;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.boot.test.context.SpringBootTest;
  10. import java.io.Serializable;
  12. @SpringBootTest
  13. class SpringAmqpTest {
  14.     @Autowired
  15.     private RabbitTemplate rabbitTemplate;
  17.     @Test
  18.     public void testSimpleQueue() {
  19.         User jack = new User("jack", 18);
  20.         rabbitTemplate.convertAndSend("testConvertMessage", jack);
  21.     }
  22. }
  23. @Data
  24. @AllArgsConstructor
  25. class User implements Serializable { //要实现Serializable接口，否则convertAndSend方法进行消息转换时会抛出异常
  26.     private String name;
  27.     private Integer age;
  28. }

  复制代码
  查看消息：
  
  
  
  消费者：
  
  1. package com.itheima.consumer.mq;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  8. import org.springframework.stereotype.Component;
  10. import java.io.Serializable;
  12. @Component
  13. @Slf4j
  14. public class SpringRabbitListener {
  15.     @RabbitListener(queues = "testConvertMessage")
  16.     public void listenerWorkQueue(User message){
  17.         log.info("消费者接收到消息：{}",message);
  18.     }
  19. }
  20. @Data
  21. @AllArgsConstructor
  22. @NoArgsConstructor
  23. class User implements Serializable {
  24.     private String name;
  25.     private Integer age;
  26. }

  复制代码
  消费者输出：
  
  
  
  
• 配置消息转换器
  发送者：
  
  1. package com.itheima.publisher;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import org.junit.jupiter.api.Test;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.boot.test.context.SpringBootTest;
  10. @SpringBootTest
  11. class SpringAmqpTest {
  12.     @Autowired
  13.     private RabbitTemplate rabbitTemplate;
  15.     @Test
  16.     public void testSimpleQueue() {
  17.         User jack = new User("jack", 18);
  18.         rabbitTemplate.convertAndSend("testConvertMessage", jack);
  19.     }
  20. }
  21. @Data
  22. @AllArgsConstructor
  23. class User {
  24.     private String name;
  25.     private Integer age;
  26. }

  复制代码
  查看消息：
  
  
  
  消费者：
  
  1. package com.itheima.consumer.mq;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  8. import org.springframework.stereotype.Component;
  11. @Component
  12. @Slf4j
  13. public class SpringRabbitListener {
  14.     @RabbitListener(queues = "testConvertMessage")
  15.     public void listenerWorkQueue(User message){ //自动将json字符串转为User独享
  16.         log.info("消费者接收到消息：{}",message);
  17.     }
  18. }
  19. @Data
  20. @AllArgsConstructor
  21. @NoArgsConstructor //消费者将消息转为User对象时，User对象一定要有空参构造器
  22. class User {
  23.     private String name;
  24.     private Integer age;
  25. }

  复制代码
  消费者输出：
  
  
  
  

消息可靠性

消息丢失三种情况：

• 发送者到MQ服务器时消息丢失
  
• MQ服务器宕机导致消息丢失
  
• MQ服务器将消息发送给消费者时消息丢失
  

发送者的可靠性

发送者重连

有的时候由于网络颠簸，可能会出现发送者连接MQ失败的情况。通过配置我们可以开启连接失败后的重连机制，默认重连机制是关闭的。

1. spring:
2.   rabbitmq:
3.     connection-timeout: 1s #设置MQ的连接超时时间，超过1秒钟还没有连上MQ则表示连接超时
4.     template:
5.       retry:
6.         enabled: true # 开启超时重试机制
7.         initial-interval: 1000ms # 失败后的初始等待时间
8.         multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 initial-interval * multiplier
9.         max-attempts: 3 # 最大重试次数

复制代码

案例演示：

• 停止MQ
  
  
  
  
• 开启重连
  
  1. spring:
  2.   rabbitmq:
  3.     host: 192.168.1.2
  4.     port: 5672
  5.     virtual-host: /hmall
  6.     username: jack
  7.     password: jack
  8.     connection-timeout: 1s
  9.     template:
  10.       retry:
  11.         enabled: true
  12.         initial-interval: 1000ms
  13.         multiplier: 1
  14.         max-attempts: 3

  复制代码
• 发送者发送消息
  
  1. package com.itheima.publisher;
  3. import org.junit.jupiter.api.Test;
  4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.boot.test.context.SpringBootTest;
  8. @SpringBootTest
  9. class SpringAmqpTest {
  10.     @Autowired
  11.     private RabbitTemplate rabbitTemplate;
  13.     @Test
  14.     public void testSimpleQueue() {
  15.         rabbitTemplate.convertAndSend("testConvertMessage", "你好");
  16.     }
  17. }

  复制代码
• 消息发送失败
  
  
  
  

注意：当网络不稳定的时候，利用重试机制可以有效进步消息发送的乐成率。不过SpringAMQP提供的重试机制是壅闭式的重试，也就是说多次重试期待的过程中，当前线程是被壅闭的，会影响业务性能，如果对于业务性能有要求，建议禁用重试机制。如果一定要利用，请合理配置期待时长和重试次数，固然也可以考虑利用异步线程来实行发送消息的代码。
发送者确认

SpringAMQP提供了Publisher Confirm和Publisher Return两种确认机制。开启确人机制后，当发送者发送消息给MQ后，MQ会返回确认结果给发送者。返回的结果有以下几种情况：

• 消息投递到了MQ，但是MQ路由失败。此时会通过PublisherReturn返回路由非常原因，然后PublisherConfirm返回ACK，告知发送者投递乐成
  
• 暂时消息投递到了MQ，而且入队乐成，PublisherConfirm返回ACK，告知投递乐成
  
• 长期消息投递到了MQ，而且入队且完成长期化，PublisherConfirm返回ACK，告知投递乐成
  
• 其它情况都会返回NACK，告知投递失败
  

开开导送者确认机制：

• 开启配置
  
  1. spring:
  2.   rabbitmq:
  3.     publisher-confirm-type: correlated # 开局publisher confirm机制，并设置confirm类型
  4.     publisher-returns: true # 开局publisher return机制

  复制代码
  这里publisher-confirm-type有三种模式可选：
  
  
  
  • none：关闭confirm机制
    
  • simple：同步壅闭期待MQ的回执消息
    
  • correlated：MQ异步回调方式返回回执消息
    
  
  
• 为RabbitTemplate配置ReturnsCallback
  每个RabbitTemplate只能配置一个ReturnsCallback，因此需要在项目启动过程中配置：
  
  
  
  
• 每次发送消息时，指定消息ID、消息ConfirmCallback
  
  
  
  

案例演示：

• 开开导送者确认配置
  
  1. spring:
  2.   rabbitmq:
  3.     host: 192.168.1.2 # RabbitMQ地址
  4.     port: 5672 # 端口
  5.     virtual-host: /hmall # 虚拟主机
  6.     username: jack # 用户名
  7.     password: jack # 密码
  8.     publisher-confirm-type: correlated # 开局publisher confirm机制，并设置confirm范例    publisher-returns: true # 开局publisher return机制

  复制代码
• 定义ReturnsCallback
  
  1. package com.itheima.publisher;
  3. import lombok.AllArgsConstructor;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  6. import org.springframework.context.annotation.Configuration;
  7. import javax.annotation.PostConstruct;
  9. @Configuration
  10. @Slf4j
  11. @AllArgsConstructor
  12. public class MqConfig {
  13.     private final RabbitTemplate rabbitTemplate;
  14.     @PostConstruct
  15.     public void init() {
  16.         rabbitTemplate.setReturnsCallback(returnedMessage -> {
  17.             log.info("监听到了消息return callback");
  18.             log.info("exchange: {}", returnedMessage.getExchange());
  19.             log.info("routingKey: {}", returnedMessage.getRoutingKey());
  20.             log.info("message：{}", returnedMessage.getMessage());
  21.             log.info("replyCode: {}", returnedMessage.getReplyCode());
  22.             log.info("replyText: {}", returnedMessage.getReplyText());
  23.         });
  24.     }
  25. }

  复制代码
• 定义ConfirmCallback并发送消息
  3.1 发送乐成
  
  1. package com.itheima.publisher;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.junit.jupiter.api.Test;
  5. import org.springframework.amqp.rabbit.connection.CorrelationData;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.boot.test.context.SpringBootTest;
  9. import org.springframework.util.concurrent.ListenableFutureCallback;
  10. import java.util.UUID;
  11. import java.util.concurrent.TimeUnit;
  13. @SpringBootTest
  14. @Slf4j
  15. class SpringAmqpTest {
  16.     @Autowired
  17.     private RabbitTemplate rabbitTemplate;
  19.     @Test
  20.     public void testConfirmCallback() {
  21.         //0. 创建CorrelationData，并设置消息ID
  22.         CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
  23.         cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
  24.             @Override
  25.             public void onFailure(Throwable ex) {
  26.                 log.error("spring amqp 处理确认结果异常", ex);
  27.             }
  28.             @Override
  29.             public void onSuccess(CorrelationData.Confirm result) {
  30.                 if (result.isAck()) {
  31.                     log.info("收到ConfirmCallback ack，消息发送成功！");
  32.                 } else {
  33.                     log.info("收到ConfirmCallback nack，消息发送失败！", result.getReason());
  34.                 }
  35.             }
  36.         });
  37.         //1. 交换机名称
  38.         String exchangeName = "hmall.direct";
  39.         //2. 消息
  40.         String message = "测试发送者确认";
  41.         //3. 发送消息
  42.         rabbitTemplate.convertAndSend(exchangeName, "blue", message, cd);
  43.         //4. 此单元测试方法执行完，main线程就结束了，因此需要睡眠2s接收回调函数
  44.         try {
  45.             TimeUnit.SECONDS.sleep(2);
  46.         } catch (InterruptedException e) {
  47.             e.printStackTrace();
  48.         }
  49.     }
  50. }

  复制代码
  
  
  3.2 发送失败-路由失败
  
  1. rabbitTemplate.convertAndSend(exchangeName, "blue22", message, cd);

  复制代码
  
  
  

注意：发送者确认机制需要发送者和MQ举行确认，会大大影响消息发送的服从，通常情况下不建议开开导送者确认机制。
MQ的可靠性

在默认情况下，RabbitMQ会将吸收到的信息保存在内存中以降低消息收发的延迟。这样会导致两个问题：

• 一旦MQ宕机，内存中的消息会丢失
  
• 内存空间有限，当消费者故障或处置惩罚过慢时，会导致消息积存，引发MQ壅闭
  

数据长期化

RabbitMQ实现数据长期化包括3个方面，设置为长期化后，重启MQ，互换机、队列、消息也不会丢失。

• 互换机长期化（新建互换机默认就是长期化）
  
  
  
  D表现长期化
  
  
  
  
• 队列长期化（新建队列默认就是长期化）
  
  
  
  
• 消息长期化（可视化界面发送消息时默认黑白长期化，SpringAmqp发送消息时默认是长期化的）
  
  
  
  

案例演示：
MQ吸收非长期化消息
发送者发送1百万条非长期化消息

发送耗时：

MQ收到了一百万条非长期化消息
注意：本测试利用的MQ是3.13.3，默认利用的是Lazy Queue模式：所有的消息直接存入磁盘，不再存储到内存，所以In memory显示1。（paged out代表的就是从内存移动到磁盘中的消息的数量，Persistent表现存入磁盘且长期化的消息的数量）

重启MQ后，一百万条非长期化消息全部丢失

MQ吸收长期化消息
发送者发送1百万条长期化消息

发送耗时：

MQ收到了一百万条长期化消息
注意：本测试利用的MQ是3.13.3，默认利用的是Lazy Queue模式：所有的消息直接存入磁盘，不再存储到内存，所以In memory显示1。（paged out代表的就是从内存移动到磁盘中的消息的数量，Persistent表现存入磁盘且长期化的消息的数量）

重启MQ后，一百万条长期化消息不会丢失

结论：
在吸收非长期化消息时，MQ收到消息后会先将消息存到内存中的队列中，队列满了之后会把先收到的消息存到磁盘中（这个行为称为paged out，paged out会导致MQ壅闭），然后再继承吸收消息，把消息存进内存中的队列中，队列满了之后再把队列中的消息存入磁盘中，以此类推。
在吸收长期化消息时，MQ会直接将消息存到磁盘中，不会等内存中的队列满了之后再将消息保存到磁盘中。
发送一万万条非长期化消息耗时：

发送一万万条长期化消息耗时：

从上面发送者发送一百万条消息的耗时来看，发送长期化消息比发送非长期化消息耗时更少（不需要paged out），而且长期化消息在MQ重启后不会丢失，所以建议发送长期化消息。
Lazy Queue

从RabbitMQ的3.6.0版本开始，就增长了Lazy Queue的概念，也就是惰性队列。
惰性队列的特性如下：

• 吸收到消息后直接存入磁盘，不再存储到内存
  
• 消费者要消费消息时才会从磁盘中读取并加载到内存（可以提前缓存部分消息到内存，最多2048条）
  

在3.12版本后，所有队列都是Lazy Queue模式，无法更改。
3.12版本之前的MQ设置Lazy Queue模式有三种方式：

• 可视化界面设置
  要设置一个队列为情性队列，只需要在声明队列时，指定x-queue-mode属性为lazy即可：
  
  
  
  
• Spring Bean方式设置
  
  
  
  
• 注解方式设置
  
  
  
  

非Lazy Queue模式+长期化消息和Lazy Queue模式+长期化消息MQ吸收消息速度对比：

消费者的可靠性

消费者确认机制

消费者确认机制(Consumer Acknowledgement）是为了确认消费者是否乐成处置惩罚消息。MQ将一条消息发送给消费者后，MQ上的这条消息处置惩罚待确认状态，当消费者处置惩罚消息结束后，应该向RabbitMO发送一个回执，告知RabbitMQ自己消息处置惩罚状态：

• ack：乐成处置惩罚消息，RabbitMQ从队列中删除该消息
  
• nack：消息处置惩罚失败，RabbitMQ需要再次投递消息
  
• reject：消息处置惩罚失败并拒绝该消息，RabbitMQ从队列中删除该消息
  

SpringAMQP已经实现了消息确认功能。并答应我们通过配置文件选择ACK处置惩罚方式，有三种方式：

• none：不处置惩罚。即消息投递给消费者后立刻ack，消息会立刻从MQ删除。非常不安全，不建议利用
  
• manual：手动模式。需要自己在业务代码中调用api，发送ack或reject，存在业务入侵，但更灵活
  
• auto：自动模式（默认模式）。SpringAMQP利用 AOP对我们的消息处置惩罚逻辑做了环绕增强，当业务正常实行时则自动返回ack。当业务出现非常时，根据非常判断返回不同结果：
  
  
  
  • 如果是业务非常（好比throw new RuntimeException），会自动返回nack
    
  • 如果是消息处置惩罚或校验非常，自动返回reject
    
  
  
  
  

案例演示-自动模式：

• 消费者配置
  
  1. spring:
  2.   rabbitmq:
  3.     host: 192.168.1.2 # RabbitMQ地址
  4.     port: 5672 # 端口
  5.     virtual-host: /hmall # 虚拟主机
  6.     username: jack # 用户名
  7.     password: jack # 密码
  8.     listener:      simple:        prefetch: 1        acknowledge-mode: auto

  复制代码
• 消费者
  
  
  
  
• 发送者
  
  
  
  查看消息状态：
  
  
  
  
  
  
  • 由于消费者抛出业务非常，所以会给MQ发送nack，然后MQ不停地向消费者投递消息
    
    
    
    
  查看消息内容
  
  
  
     
  
  
  
  • 查看队列中的消息，提示队列是空的，所以得出结论：待确认的消息不保存在队列中
    
  
  

案例演示-手动模式：

• 消费者配置
  
  1. spring:
  2.   rabbitmq:
  3.     host: 192.168.1.2 # RabbitMQ地址
  4.     port: 5672 # 端口
  5.     virtual-host: /hmall # 虚拟主机
  6.     username: jack # 用户名
  7.     password: jack # 密码
  8.     listener:      simple:        prefetch: 1        acknowledge-mode: manual

  复制代码
• 消费者
  
  
  
  
• 发送者
  3.1 发送者发送ackxxxx
  
  
  
  步伐会运行到消费者在21行的断点处，消费者输出
  
  
  
  查看消息状态
  
  
  
  放行消费者21行断点，查看消息状态
  
  
  
  3.2 发送者发送nackxxxx，步伐会运行到消费者在26行的断点处，消费者输出
  
  
  
  查看消息状态
  
  
  
  放行断点，消费者输出
  
  
  
  查看消息状态，消息又被重新放回了队列，而且MQ又将消息投递给了消费者
  
  
  
  取消断点26行断点，消费者不停地输出，阐明MQ不停地向消费者举行消息重投
  
  
  
  查看消息状态
  
  
  
  停掉消费者进程，查看消息状态
  
  
  
  3.3 发送者发送xxxx，消费者停在30行的断点处，消费者输出
  
  
  
  查看消息状态
  
  
  
  放行断点，查看消息状态
  
  
  
  

失败重试机制

SpringAMQP提供了消费者失败重试机制，在消费者出现非常时利用当地重试，而不是无穷的requeue到mq。我们可以通过在application.yaml文件中添加配置来开启重试机制：

案例演示：

• 消费者配置
  

1. spring:
2.   rabbitmq:
3.     host: 192.168.1.2 # RabbitMQ地址
4.     port: 5672 # 端口
5.     virtual-host: /hmall # 虚拟主机
6.     username: jack # 用户名
7.     password: jack # 密码
8.     listener:      simple:        prefetch: 1        acknowledge-mode: auto        retry:          enabled: true          initial-interval: 1000ms          multiplier: 1          max-attempts: 3          stateless: true

复制代码

• 消费者
  
  
  
  
• 发送者
  
  
  
  
• 消费者输出
  
  
  
  
• 查看消息状态
  
  
  
  

在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecoverer接口来处置惩罚，它包含三种不同的实现：

• RejectAndDontRequeueRecoverer（默认）：重试耗尽后，给MQ返回reject，MQ收到reject后会将消息丢弃。
  
• ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队。
  
• RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的互换机。
  

将失败处置惩罚策略改为RepublishMessageRecoverer：

• 首先，定义吸收失败消息的互换机、队列及其绑定关系。
  
• 然后，定义RepublishMessageRecoverer：
  
  
  
  

案例演示：

• 定义吸收失败消息的互换机、队列、绑定关系、RepublishMessageRecoverer
  

• 消费者
  
  
  
  
• 消费者输出
  
  
  
  
• 查看error.queue上的消息
  
  
  
  

业务幂等性

幂等是一个数学概念，用函数表达式来形貌是这样的：f(x)=f(f(x))，例如求绝对值的函数。在步伐开发中，则是指同一个业务，实行一次或多次对业务状态的影响是一致的。

消除非幂等性的手段：

• 唯一消息id
  
  
  
  

案例演示：

• 配置消息转换器
  
  
  
  
• 发送者发送消息
  
  
  
  
• 查看消息
  
  
  
  
• 消费者利用Message吸收
  

• 业务判断
  
  
  
  

延迟消息

延迟消息：发送者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息。
延迟任务：设置在一定时间之后才实行的任务

死信互换机

当一个队列中的消息满意下列情况之一时，就会成为死信 (dead letter）

• 消费者利用basic.reject或 basic.nack声明消费失败，而且消息的requeue参数设置为false
  
• 消息是一个过期消息（到达了队列或消息本身设置的过期时间），超时无人消费
  
• 要投递的队列消息堆积满了，最早的消息可能成为死信
  

如果队列通过dead-letter-exchange属性指定了一个互换机，那么该队列中的死信就会投递到这个互换机中。这个互换机称为死信互换机(Dead Letter Exchange，简称DLX）

案例演示：

• 消费者中定义互换机和队列，并监听
  
  
  
  
• 定义互换机和队列，并将互换机dlx.direct声明为死信互换机，并与队列normal.queue绑定。
  
  
  
  
• 查看队列、互换机、绑定关系
  
  
  
  
  
  
  
  
  
  
• 发送者，发送消息时设置消息的殒命时间
  
  
  
  
• 大约10s后消费者收到消息
  
  
  
  
  
  
  
• 阐明
  向互换机normal.direct中投递消息M，指定routingKey为hi，消息M的殒命时间设置为10s，消息M会被路由到队列normal.queue中，由于队列normal.queue没有消费者监听，可巧队列normal.queue绑定了死信互换机dlx.direct，所以投递到队列normal.queue的消息M殒命后，会被转投到死信互换机dlx.direct中，由于指定的routingKey为hi，所以死信互换机dlx.direct会将消息M路由到队列dlx.queue中，而队列dlx.queue有消费者C监听，所以消费者C会消费消息M。这就实现了消息M延迟10秒后被消费。
  

延迟消息插件

利用死信队列可以实现延迟消息，但这种方法过于繁琐。为了简化这一过程，RabbitMQ的官方推出了一款插件，该插件原生支持延迟消息功能。该插件的运作原理是计划了一种特别的互换机，当消息投递到这种互换机时，它能够暂存一段时间，直到到达设定的延迟时间后再将消息投递到相应的队列。这种计划大大简化了延迟消息的处置惩罚过程，进步了系统的服从和可靠性。
下载：
下载地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases
选择下载的版本，插件的版本要和RabbitMQ的版本保持一致

• 查看RabbitMQ的版本：
  
  1. docker run -d --name containerId -p 5672:5672 rabbitmq:3-management

  复制代码
  
  
  
• 选择3.13.x版本的插件
  
  
  
  

安装：

• 将插件复制到RabbitMQ容器中
  
  1. docker cp 插件路径 容器ID或名称:/plugins/

  复制代码
• 安装
  
  1. docker exec -it 容器ID或名称 rabbitmq-plugins enable rabbitmq_delayed_message_exchange

  复制代码
  
  
  

利用：

• 创建延迟互换机，三种方式
  
  
  
  • 图形化界面操纵
    
    
    
    
  • 注解方式
    
    
    
    
  • SpringBean方式
    
    
    
    
  
  
• 查看延迟互换机
  
  
  
  
• 发送消息时需要通过消息头x-delay来设置过期时间
  
  
  
  
• 消费者大约10s后收到消息
  
  
  
  
  
  
  

本文参考文档

https://b11et3un53m.feishu.cn/wiki/A9SawKUxsikJ6dk3icacVWb4n3g
https://blog.csdn.net/karry_zzj/article/details/119513541
https://blog.csdn.net/weixin_42050545/article/details/121487823

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。