谷粒商城高级篇 (十九) --------- 消息队列

矫情吗；* 2024-04-06 11:16 41阅读 0赞

#### 目录 ####

*  一、概述
 *  二、应用
 *  三、RabbitMQ 概念
 *  四、安装 RabbitMQ
 *  五、RabbitMQ 运行机制
 *   *  Exchange 类型
 *  六、RabbitMQ 整合
 *  七、RabbitMQ 消息确认机制
 *   *  1. ConfirmCallback
     *  2. ReturnCallback
     *  3. Ack 消息确认机制
 *  八、RabbitMQ 延时队列 (实现定时任务)
 *   *  1. 消息的TTL (Time To Live)
     *  2. Dead Letter Exchanges (DLX)
     *  3. SpringBoot 中使用延时队列
 *  九、如何保证消息可靠性
 *   *  1. 消息丢失
     *  2. 消息重复
     *  3. 消息积压
 *  十、三种 MQ 之间的对比

--------------------

## 一、概述 ##

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

**消息服务中两个重要概念：**

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

**消息队列主要有两种形式的目的地：**

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

**点对点式：**

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

**发布订阅式：**

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

**JMS (Java Message Service) JAVA消息服务：**

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

**AMQP (Advanced Message Queuing Protocol)**

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

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

**Spring 支持**

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

**Spring Boot自动配置**

*  JmsAutoConfiguration
 *  RabbitAutoConfiguration

**市面的MQ产品**

*  ActiveMQ、RabbitMQ、RocketMQ、Kafka

## 二、应用 ##

消息队列主要用作 异步、解耦与流控

**异步处理：**

![在这里插入图片描述][512a3f2646124bbeb33368ad6e460904.png]

**应用解耦：**

![在这里插入图片描述][91946525f68c49a792b70522781d10a4.png]

**流量控制：**

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

## 三、RabbitMQ 概念 ##

![在这里插入图片描述][1d580e7ebac24b1cb6905867914af80b.png]

**RabbitMQ简介**

RabbitMQ是一个由erlang开发的AMQP(Advanved Message Queue Protocol)的开源实现。核心概念  
Message 消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key (路由键)、priority (相对于其他消息的优先权)、delivery-mode (指出该消息可能需要持久性存储) 等。

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

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

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

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

**Connection**  
网络连接，比如一个TCP连接。

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

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

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

**Broker**  
表示消息队列服务器实体

![在这里插入图片描述][87291bab3f1043328e8e3bd1252d003f.png]

## 四、安装 RabbitMQ ##

在 linux 上输入如下指令：

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

如 docker 没有 rabbitmq 镜像会在第一次执行此行指令时自动拉取 rabbitmq 镜像。。。

![在这里插入图片描述][1c7a66bc9f0d411784ab266302f72a5e.png]

端口说明：

*  **4369、25672：** Erlang 发现 & 集群端口
 *  **5672、5671：** AMQP端口
 *  **15672：** web管理后台端口
 *  **61613、61614：** STOMP协议端口
 *  **1883、8883：** MQTT协议端口

[https://www.rabbitmq.com/networking.html][https_www.rabbitmq.com_networking.html]

![在这里插入图片描述][89b31e8da7ad43969b4e148f90238cef.png]  
安装完之后，我们可以进入其后台管理页面。。。

地址为: `服务器ip:15672`，登录的账号密码均为 guest

![在这里插入图片描述][ea26167ce85944efbcaab1789d1c92e3.png]  
![在这里插入图片描述][3ff18a0b6787401892bf6deb2aa6e83f.png]

**添加 direct 类型的交换机**

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

![在这里插入图片描述][5b425015f20e40d89afc5fc5e7e19b83.png]

**添加队列**

![在这里插入图片描述][6deef089b03d4268a445ddb22bddcad0.png]

**将交换机与队列进行绑定**

![在这里插入图片描述][0782b7129f0545a4bd2ef7ad58a2f110.png]

![在这里插入图片描述][3c0cbda462f840088b1d7a83038e5756.png]  
![在这里插入图片描述][5975d81b64084907b08f06fb14b2a94d.png]  
**发送消息**

![在这里插入图片描述][f00b59c68e904ae38e031b048f68fef4.png]  
![在这里插入图片描述][ae843aba835c4ab2b33a4e94c4f71fb6.png]

**查看消息**

![在这里插入图片描述][f6c4f8dbcf1e41bc911859c54617c42d.png]  
![在这里插入图片描述][7a00a3f54ee3450f9c575112ddad9c5c.png]  
![在这里插入图片描述][4fca569288944862ae60bcea4cd627d8.png]  
更多测试我们就不再详细展示。。。

## 五、RabbitMQ 运行机制 ##

AMQP 中消息的路由过程和 Java 开发者熟悉的 JMS 存在一些差别，AMQP 中增加了 Exchange 和  
Binding 的角色。生产者把消息发布到 Exchange 上，消息最终到达队列并被消费者接收，而 Binding 决定交换器的消息应该发送到那个队列。

![在这里插入图片描述][9d04b0516ee845c5b7b14054947143a6.png]

### Exchange 类型 ###

Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键，headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了，所以直接看另外三种类型：

消息中的路由键 (routingkey) 如果和 Binding 中的 binding key一致，交换器就将消息发到对应的队列中。路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为 “dog”，则只转发 routingkey 标记为 “dog” 的消息，不会转发 “dog.puppy” ，也不会转发 “dog.guard” 等等。它是完全匹配、单播的模式。

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

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

![在这里插入图片描述][22e717f0126d4b358bcc64767dd9ebf1.png]

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

![在这里插入图片描述][4bc3260f157345da876090aae470c03e.png]

## 六、RabbitMQ 整合 ##

步骤：

使用RabbitMQ
     * 1、引入amqp场景；RabbitAutoConfiguration 就会自动生效
     *
     * 2、给容器中自动配置了
     *      RabbitTemplate、AmqpAdmin、CachingConnectionFactory、RabbitMessagingTemplate;
     *      所有的属性都是 spring.rabbitmq
     *      @ConfigurationProperties(prefix = "spring.rabbitmq")
     *      public class RabbitProperties
     *
     * 3、给配置文件中配置 spring.rabbitmq 信息
     * 4、@EnableRabbit: @EnableXxxxx；开启功能
     * 5、监听消息：使用@RabbitListener；必须有@EnableRabbit
     *    @RabbitListener: 类+方法上（监听哪些队列即可）
     *    @RabbitHandler：标在方法上（重载区分不同的消息）

**A、引入 spring-boot-starter-amqp 依赖**

![在这里插入图片描述][92a6f21698924cafb13e6adabc754a13.png]

**B、application.yml配置**

![在这里插入图片描述][310c3016159640d49a1227276fc5a55c.png]

# RabbitMQ配置
    spring.rabbitmq.host=192.168.38.130
    spring.rabbitmq.port=5672
    
    # 虚拟主机配置
    spring.rabbitmq.virtual-host=/
    
    # 开启发送端消息抵达Broker确认
    spring.rabbitmq.publisher-confirms=true
    
    # 开启发送端消息抵达Queue确认
    spring.rabbitmq.publisher-returns=true
    
    # 只要消息抵达Queue，就会异步发送优先回调returnfirm
    spring.rabbitmq.template.mandatory=true
    
    # 手动ack消息，不使用默认的消费端确认
    spring.rabbitmq.listener.simple.acknowledge-mode=manual

**C、添加 RabbitMQ 功能的注解**

![在这里插入图片描述][2851a78a208049698edaf1b37f6b3c0a.png]

**D、测试 RabbitMQ**

package com.fancy.gulimall.order;
    
    import com.fancy.gulimall.order.entity.OrderEntity;
    import com.fancy.gulimall.order.entity.OrderReturnReasonEntity;
    import lombok.extern.slf4j.Slf4j;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.amqp.core.AmqpAdmin;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import java.util.Date;
    import java.util.UUID;
    
    @SpringBootTest
    @Slf4j
    @RunWith(SpringRunner.class)
    class GulimallOrderApplicationTests {
        
       
        //1. AmqpAdmin：管理组件
        @Autowired
        AmqpAdmin amqpAdmin;
        
        // 2. RabbitTemplate：消息发送处理组件
        @Autowired
        RabbitTemplate  rabbitTemplate;
    
        @Test
        public void sendMessageTest() {
        
            // 1. 发送消息, 如果发送的消息是个对象, 我们会使用序列化机制, 将对象写出去。对象必须实现 Serializable
            String msg = "Hello World";
    
            // 2. 发送的对象类型的转换消息, 可以转换成一个 json
            for (int i = 0; i < 10; i ++ ) {
        
                if (i % 2 == 0) {
        
                    OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
                    reasonEntity.setId(1L);
                    reasonEntity.setCreateTime(new Date());
                    reasonEntity.setName("哈哈-" + i);
                    rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", reasonEntity, new CorrelationData(UUID.randomUUID().toString()));
                } else {
        
                    OrderEntity entity = new OrderEntity();
                    entity.setOrderSn(UUID.randomUUID().toString());
                    rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", entity,new CorrelationData(UUID.randomUUID().toString()));
                }
    
                log.info("消息发送完成{}");
            }
        }
    
        /**
         * 1、如何创建Exchange[hello-java-exchange]、Queue、Binding
         * 1）、使用 AmqpAdmin 进行创建
         * 2、如何收发消息
         */
        @Test
        public void createExchange() {
        
            //amqpAdmin
            //Exchange
            /**
             * DirectExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments)
             */
            DirectExchange directExchange = new DirectExchange("hello-java-exchange", true, false);
            amqpAdmin.declareExchange(directExchange);
            log.info("Exchange[{}]创建成功", "hello-java-exchange");
    
        }
    
        @Test
        public void createQueue() {
        
            //public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
            Queue queue = new Queue("hello-java-queue", true, false, false);
            amqpAdmin.declareQueue(queue);
            log.info("Queue[{}]创建成功", "hello-java-queue");
        }
    
        @Test
        public void createBinding() {
        
            //(String destination【目的地】,
            // DestinationType destinationType【目的地类型】,
            // String exchange【交换机】,
            // String routingKey【路由键】,
            //Map<String, Object> 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");
        }
    
    }

**E、测试接收消息**

在上面一段代码中我们编写了在测试类中发送消息的代码，接下来我们测试来接收消息。。。，接受消息我们主要在类或方法上加入 @RabbitListener 或· @RabbitHandler 来执行

例如：

/**
    * queues：声明需要监听的所有队列
    *
    * org.springframework.amqp.core.Message
    *
    * 参数可以写一下类型
    * 1、Message message：原生消息详细信息。头+体
    * 2、T<发送的消息的类型> OrderReturnReasonEntity content；
    * 3、Channel channel：当前传输数据的通道
    *
    * Queue：可以很多人都来监听。只要收到消息，队列删除消息，而且只能有一个收到此消息
    * 场景：
    *    1）、订单服务启动多个；同一个消息，只能有一个客户端收到
    *    2)、 只有一个消息完全处理完，方法运行结束，我们就可以接收到下一个消息
    */
    //    @RabbitListener(queues = {"hello-java-queue"})
    @RabbitHandler
    public void recieveMessage(Message message,
                              OrderReturnReasonEntity content,
                              Channel channel) throws InterruptedException {
        
       //{"id":1,"name":"哈哈","sort":null,"status":null,"createTime":1581144531744}
       System.out.println("接收到消息..."+content);
       byte[] body = message.getBody();
       //消息头属性信息
       MessageProperties properties = message.getMessageProperties();
    //        Thread.sleep(3000);
       System.out.println("消息处理完成=>"+content.getName());
       //channel内按顺序自增的。
       long deliveryTag = message.getMessageProperties().getDeliveryTag();
       System.out.println("deliveryTag==>"+deliveryTag);
    
       //签收货物,非批量模式
       try {
        
           if(deliveryTag%2 == 0){
        
               //收货
               channel.basicAck(deliveryTag,false);
               System.out.println("签收了货物..."+deliveryTag);
           }else {
        
               //退货 requeue=false 丢弃  requeue=true 发回服务器，服务器重新入队。
               //long deliveryTag, boolean multiple, boolean requeue
               //签收了货物...6
               channel.basicNack(deliveryTag,false,true);
               //long deliveryTag, boolean requeue
    //                channel.basicReject();
               System.out.println("没有签收了货物..."+deliveryTag);
           }
    
       }catch (Exception e){
        
           //网络中断
       }
    
    }
    
    @RabbitHandler
    public void recieveMessage2(OrderEntity content) throws InterruptedException {
        
       //{"id":1,"name":"哈哈","sort":null,"status":null,"createTime":1581144531744}
       System.out.println("接收到消息..."+content);
    }

`@RabbitListener(queues = {"hello-java-queue"})` 可以加到类上面去，然后方法上标注 `@RabbitHandler` 去处理。。。，区别就是 `@RabbitHandler` 标注在方法上可以重载区分不同的消息

## 七、RabbitMQ 消息确认机制 ##

RabbitMQ 消息确认机制保证消息的可靠抵达，可以使用以下几种方式。。。

*  保证消息不丢失，可靠抵达，可以使用事务消息，性能下降250倍，为此引入确认机制
 *  publisher confirmCallback 确认模式
 *  publisher returnCallback 未投递到 queue 退回模式
 *  consumer ack机制

![在这里插入图片描述][5a2a9fd09fc04d74979b13706a22d62b.png]

### 1. ConfirmCallback ###

表示服务器收到消息就回调

*  spring.rabbitmq.publisher-confirms=true  
    \- 设置确认回调ConfirmCallback  
    在配置文件中加入：

spring.rabbitmq.publisher-confirms=true

在创建 connectionFactory 的时候设置 PublisherConfirms (true) 选项，开启 ConfirmCallback ，因此我们可以定制 RabbitTemplate ，在 config 包下，创建 MyRabbitConfig 配置类。。。，设置确认回调，需要编写如下方法

//设置确认回调
    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        
    
        /**
         *
         * 1、只要消息抵达Broker就ack=true
         * @param correlationData 当前消息的唯一关联数据（这个是消息的唯一id）
         * @param ack  消息是否成功收到
         * @param cause 失败的原因
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        
    
            /**
             * 1、做好消息确认机制（pulisher，consumer【手动ack】）
             * 2、每一个发送的消息都在数据库做好记录。定期将失败的消息再次发送一遍
             */
            //服务器收到了；
            //修改消息的状态
            System.out.println("confirm...correlationData["+correlationData+"]==>ack["+ack+"]==>cause["+cause+"]");
        }
    });

说明：

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

### 2. ReturnCallback ###

消息正确抵达队列进行回调，有以下两个步骤：

1、在 application.properties 中进行配置：

spring.rabbitmq.publisher-returns=true
    spring.rabbitmq.template.mandatory=true

2、设置确认回调ReturnCallback

//设置消息抵达队列的确认回调
    rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
        
        /**
         * 只要消息没有投递给指定的队列，就触发这个失败回调
         * @param message   投递失败的消息详细信息
         * @param replyCode 回复的状态码
         * @param replyText 回复的文本内容
         * @param exchange  当时这个消息发给哪个交换机
         * @param routingKey 当时这个消息用哪个路由键
         */
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        
            //报错误了。修改数据库当前消息的状态->错误。
            System.out.println("Fail Message["+message+"]==>replyCode["+replyCode+"]==>replyText["+replyText+"]===>exchange["+exchange+"]===>routingKey["+routingKey+"]");
        }
    });

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

MyRabbitConifg 完整配置如下：

package com.fancy.gulimall.order.config;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Primary;
    
    import javax.annotation.PostConstruct;
    
    @Configuration
    public class MyRabbitConfig {
        
    
    //
    //    @Autowired
        RabbitTemplate rabbitTemplate;
    
    
    //    public MyRabbitConfig(RabbitTemplate rabbitTemplate){
        
    //        this.rabbitTemplate = rabbitTemplate;
    //        initRabbitTemplate();
    //    }
        //TODO
        @Primary
        @Bean
        public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        
            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
            this.rabbitTemplate = rabbitTemplate;
            rabbitTemplate.setMessageConverter(messageConverter());
            initRabbitTemplate();
            return rabbitTemplate;
        }
    
        /**
         * 使用JSON序列化机制，进行消息转换
         */
        @Bean
        public MessageConverter messageConverter(){
        
    
            return new Jackson2JsonMessageConverter();
        }
    
        /**
         * 定制RabbitTemplate
         * 1、服务器收到消息就回调
         *      1、spring.rabbitmq.publisher-confirms=true
         *      2、设置确认回调ConfirmCallback
         * 2、消息正确抵达队列进行回调
         *      1、 spring.rabbitmq.publisher-returns=true
         *          spring.rabbitmq.template.mandatory=true
         *      2、设置确认回调ReturnCallback
         *
         * 3、消费端确认（保证每个消息被正确消费，此时才可以broker删除这个消息）。
         *      spring.rabbitmq.listener.simple.acknowledge-mode=manual 手动签收
         *      1、默认是自动确认的，只要消息接收到，客户端会自动确认，服务端就会移除这个消息
         *          问题：
         *              我们收到很多消息，自动回复给服务器ack，只有一个消息处理成功，宕机了。就会发生消息丢失；
         *              消费者手动确认模式。只要我们没有明确告诉MQ，货物被签收。没有Ack，
         *                  消息就一直是unacked状态。即使Consumer宕机。消息不会丢失，会重新变为Ready，下一次有新的Consumer连接进来就发给他
         *      2、如何签收:
         *          channel.basicAck(deliveryTag,false);签收；业务成功完成就应该签收
         *          channel.basicNack(deliveryTag,false,true);拒签；业务失败，拒签
         */
    //    @PostConstruct //MyRabbitConfig对象创建完成以后，执行这个方法
        public void initRabbitTemplate(){
        
            //设置确认回调
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        
    
                /**
                 *
                 * 1、只要消息抵达Broker就ack=true
                 * @param correlationData 当前消息的唯一关联数据（这个是消息的唯一id）
                 * @param ack  消息是否成功收到
                 * @param cause 失败的原因
                 */
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        
    
                    /**
                     * 1、做好消息确认机制（pulisher，consumer【手动ack】）
                     * 2、每一个发送的消息都在数据库做好记录。定期将失败的消息再次发送一遍
                     */
                    //服务器收到了；
                    //修改消息的状态
                    System.out.println("confirm...correlationData["+correlationData+"]==>ack["+ack+"]==>cause["+cause+"]");
                }
            });
    
            //设置消息抵达队列的确认回调
            rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
        
                /**
                 * 只要消息没有投递给指定的队列，就触发这个失败回调
                 * @param message   投递失败的消息详细信息
                 * @param replyCode 回复的状态码
                 * @param replyText 回复的文本内容
                 * @param exchange  当时这个消息发给哪个交换机
                 * @param routingKey 当时这个消息用哪个路由键
                 */
                @Override
                public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        
                    //报错误了。修改数据库当前消息的状态->错误。
                    System.out.println("Fail Message["+message+"]==>replyCode["+replyCode+"]==>replyText["+replyText+"]===>exchange["+exchange+"]===>routingKey["+routingKey+"]");
                }
            });
        }
    }

### 3. Ack 消息确认机制 ###

自动回复一旦服务宕机就会发生丢失，因此我们可以采取手动确认机制。。。

在配置文件中加入：

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

如何签收消息：

channel.basicAck(deliveryTag,false)：签收，业务成功完成就应该签收
    channel.basicNack(deliveryTag,false,true)，拒签；业务失败，拒签

*  消费者获取到消息，成功处理，可以回复 Ack 给 Broker
 *  basic.ack 用于肯定确认，broker 将移除此消息
 *  basic.nack 用于否定确认，可以指定 broker 是否丢弃此消息，可以批量
 *  basic.reject 用于否定确认，同上，但不能批量
 *  默认自动ack，消息被消费者收到，就会从broker的queue中移除
 *  queue无消费者，消息依然会被存储，直到消费者消费
 *  消费者收到消息，默认会自动ack。但是如果无法确定此消息是否被处理完成，或者成功处理。我们可以开启手动ack模式
 *  消息处理成功，ack()，接受下一个消息，此消息broker就会移除
 *  消息处理失败，nack()/reject()，重新发送给其他人进行处理，或者容错处理后ack
 *  消息一直没有调用ack/nack方法，broker认为此消息正在被处理，不会投递给别人，此时客户端断开，消息不会被broker移除，会投递给别人

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

## 八、RabbitMQ 延时队列 (实现定时任务) ##

**场景：**

比如未付款订单，超过一定时间后，系统自动取消订单并释放占有物品。

常用解决方案：spring的 schedule 定时任务轮询数据库

**缺点：**

消耗系统内存、增加了数据库的压力、存在较大的时间误差

解决：RabbitMQ 的消息 TTL 和死信 Exchange 结合

### 1. 消息的TTL (Time To Live) ###

消息的 TTL 就是消息的存活时间。 RabbitMQ 可以对队列和消息分别设置 TTL。

对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。

如果队列设置了，消息也设置了，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息死亡的时间有可能不一样(不同的队列设置)。

这里单讲单个消息的TTL，因为它才是实现延迟任务的关键。可以通过设置消息的 expiration 字段或者 x-message-ttl 属性来设置时间，两者是一样的效果。

### 2. Dead Letter Exchanges (DLX) ###

**什么是死信**

一个消息被 Consumer 拒收了，并且 reject 方法的参数里 requeue 是 false。也就是说不会被再次放在队列里，被其他消费者使用。`(basic.reject / basic.nack)requeue=false`

一个消息在满足如下条件下，会进死信路由，记住这里是路由而不是队列，一个路由可以对应很多队列。

*  上面的消息的TTL到了，消息过期了。
 *  队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上

Dead Letter Exchange 其实就是一种普通的 exchange，和创建其他 exchange 没有两样。只是在某一个设置 Dead Letter Exchange的队列中有消息过期了，会自动触发消息的转发，发送到 Dead Letter Exchange 中去。

我们既可以控制消息在一段时间后变成死信，又可以控制变成死信的消息被路由到某一个指定的交换机，结合二者，其实就可以实现一个延时队列。

手动 ack & 异常消息统一放在一个队列处理建议的两种方式：

*  catch异常后，手动发送到指定队列，然后使用 channel 给 RabbitMQ 确认消息已消费
 *  给Queue绑定死信队列，使用 nack (requque为false) 确认消息消费失败

![在这里插入图片描述][057d8a2ad7cf44dc879eba3da89e7721.png]  
**实现方式**

方式一：  
![在这里插入图片描述][7d3922c77a5a4d75ac3f106a4ace1511.png]

方式二：  
![在这里插入图片描述][a5cb94e4a4e24116ad25c6a195cffca1.png]

业务中实现：

![在这里插入图片描述][702ffeb4faa24dc2ade008a1d35d787a.png]  
![在这里插入图片描述][4f7d039e78c6495f97cb0b6c58c69c72.png]

### 3. SpringBoot 中使用延时队列 ###

*  Queue、Exchange、Binding可以 @Bean 进去
 *  监听消息的方法可以有三种参数 (不分数量，顺序) `Object content, Message message, Channel channel`
 *  channel 可以用来拒绝消息，否则自动 ack

Queue、Echange、Binding 可以通过配置类进行创建 @Bean 注解会将其自动进行注入创建。。。

package com.fancy.gulimall.order.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    public class MyMQConfig {
        
    
    
    
        //@Bean Binding，Queue，Exchange
    
        /**
         * 容器中的 Binding，Queue，Exchange 都会自动创建（RabbitMQ没有的情况）
         * RabbitMQ 只要有。@Bean声明属性发生变化也不会覆盖
         * @return
         */
        @Bean
        public Queue orderDelayQueue() {
        
    
            Map<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);
            //String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
            Queue queue = new Queue("order.delay.queue", true, false, false,arguments);
            return queue;
        }
    
        @Bean
        public Queue orderReleaseOrderQueue() {
        
            Queue queue = new Queue("order.release.order.queue", true, false, false);
            return queue;
        }
    
        @Bean
        public Exchange orderEventExchange() {
        
            //String name, boolean durable, boolean autoDelete, Map<String, Object> arguments
           return new TopicExchange("order-event-exchange",true,false);
        }
    
        @Bean
        public Binding orderCreateOrderBingding() {
        
            //String destination, 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);
    
        }
    
        @Bean
        public Binding orderReleaseOrderBingding() {
        
            return new Binding("order.release.order.queue",
                    Binding.DestinationType.QUEUE,
                    "order-event-exchange",
                    "order.release.order",
                    null);
        }
    
    
        /**
         * 订单释放直接和库存释放进行绑定
         * @return
         */
        @Bean
        public Binding orderReleaseOtherBingding() {
        
            return new Binding("stock.release.stock.queue",
                    Binding.DestinationType.QUEUE,
                    "order-event-exchange",
                    "order.release.other.#",
                    null);
        }
    
    
        @Bean
        public Queue orderSeckillOrderQueue(){
        
            //String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
            return new Queue("order.seckill.order.queue",true,false,false);
        }
    
        @Bean
        public Binding orderSeckillOrderQueueBinding(){
        
            /**
             * String destination, DestinationType destinationType, String exchange, String routingKey,
             * 			Map<String, Object> arguments
             */
            return new Binding("order.seckill.order.queue",
                    Binding.DestinationType.QUEUE,
                    "order-event-exchange",
                    "order.seckill.order",
                    null);
        }
    
    }

## 九、如何保证消息可靠性 ##

### 1. 消息丢失 ###

**场景：**

消息发送出去，由于网络问题没有抵达服务器。

**解决：**  
、

*  做好容错方法 (try-catch)，发送消息可能会网络失败，失败后要有重试机制，可记录到数据库，采用定期扫描重发的方式。
 *  做好日志记录，每个消息状态是否都被服务器收到都应该记录
 *  做好定期重发，如果消息没有发送成功，定期去数据库扫描未成功的消息进行重发

**场景：**

消息抵达 Broker，Broker 要将消息写入磁盘 (持久化) 才算成功。此时Broker尚未持久化完成，宕机。

**解决：**

*  publisher也必须加入确认回调机制，确认成功的消息，修改数据库消息状态。

**场景：**

自动ACK的状态下。消费者收到消息，但没来得及消息然后宕机。

**解决：**

*  一定开启手动ACK，消费成功才移除，失败或者没来得及处理就noAck并重新入队

### 2. 消息重复 ###

**场景：**

消息消费成功，事务已经提交，ack时，机器宕机。导致没有ack成功，Broker 的消息重新由 unack 变为 ready，并发送给其他消费者。

消息消费失败，由于重试机制，自动又将消息发送出去。

成功消费，ack时宕机，消息由unack变为ready，Broker又重新发送

**解决：**

*  消费者的业务消费接口应该设计为幂等性的。比如扣库存有工作单的状态标志
 *  使用防重表 (redis/mysql)，发送消息每一个都有业务的唯一标识，处理过就不用处理
 *  RabbitMQ 的每一个消息都有 redelivered 字段，可以获取是否是被重新投递过来的，而不是第一次投递过来的

### 3. 消息积压 ###

**场景：**

消费者宕机积压

消费者消费能力不足积压

发送者发送流量太大

**解决：**

*  上线更多的消费者，进行正常消费
 *  上线专门的队列消费服务，将消息先批量取出来，记录数据库，离线慢慢处理

## 十、三种 MQ 之间的对比 ##

![在这里插入图片描述][509326dbd2824378af5607ded7fc7017.jpeg_pic_center]

[d132e3043ee84b1190f44f3d08354b09.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/faf34e6d1dbd4be39cf52545ce3d70bd.png
[512a3f2646124bbeb33368ad6e460904.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/d55af7955718447cb49d6a5e93533cc2.png
[91946525f68c49a792b70522781d10a4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/4a3d22568bfa4e22add7ab4e33691daf.png
[ffc273d42fc24e039ed2f8f19c17a46f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/06196aae25c34da88d8add1a825fc196.png
[1d580e7ebac24b1cb6905867914af80b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/dd9e7584e8ae4517a1ef989637ddbfd8.png
[87291bab3f1043328e8e3bd1252d003f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/5b9b76198be64319b5fce38c9750a252.png
[1c7a66bc9f0d411784ab266302f72a5e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/b87a483b7f474ab38040c87242ca2c8d.png
[https_www.rabbitmq.com_networking.html]: https://www.rabbitmq.com/networking.html
[89b31e8da7ad43969b4e148f90238cef.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/1ebc74f3fb11426b8d61911ea5af854e.png
[ea26167ce85944efbcaab1789d1c92e3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/982494bfc7f145ce8c5f992c32fa45b3.png
[3ff18a0b6787401892bf6deb2aa6e83f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/59a967395f584e5dbf93cf703df6cfd5.png
[bbf1e3da034b46379f167474237fce21.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/389ff190524e4b85b68e8ed4a9d545e0.png
[5b425015f20e40d89afc5fc5e7e19b83.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/dc3ee59bffd24ab79227c20ef706fd76.png
[6deef089b03d4268a445ddb22bddcad0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/bd07928963cb4cd69557241043d2e951.png
[0782b7129f0545a4bd2ef7ad58a2f110.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/870d12af16fc4676b521a9734cabd8e6.png
[3c0cbda462f840088b1d7a83038e5756.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/75664186d72940c4b89e4ae6f3c933c9.png
[5975d81b64084907b08f06fb14b2a94d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/60bd1b40822a4fb2b187ae42f479df50.png
[f00b59c68e904ae38e031b048f68fef4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/ada3ffd5057043f8bb58dbf82adbf049.png
[ae843aba835c4ab2b33a4e94c4f71fb6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/3bcbf28fd3b14c9780626e5bd2c79548.png
[f6c4f8dbcf1e41bc911859c54617c42d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/f4de33072f584d9a87020874e0fc9f2d.png
[7a00a3f54ee3450f9c575112ddad9c5c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/8f65d5ca482648ae95cb437b0de83353.png
[4fca569288944862ae60bcea4cd627d8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/a0e6f5570c5043cdb446e69775f4af4b.png
[9d04b0516ee845c5b7b14054947143a6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/6f57a0d369044dc2b356186c156fba46.png
[dbf6b8694b5842a3bff6acf5fda02a28.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/529c9a0d89a34c24ac02f8cf2ae04225.png
[22e717f0126d4b358bcc64767dd9ebf1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/e8e314578c8246db8f041dadf9a92920.png
[4bc3260f157345da876090aae470c03e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/9e142e3b60034d4cb3b8ce1ae2ecd11b.png
[92a6f21698924cafb13e6adabc754a13.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/d638d32986084b1d9b54f785975240d3.png
[310c3016159640d49a1227276fc5a55c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/e4dee7b79bc641a797e88fbfbef066e1.png
[2851a78a208049698edaf1b37f6b3c0a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/391592ce578d4df594b413e4cdf81714.png
[5a2a9fd09fc04d74979b13706a22d62b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/84e1dd0a2cd8474993f9f9d9fbbe02f8.png
[f9c811ef7aac4b0ab5d630fab553e29e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/e4543a96ea6b445daeb9feb6fe4f7e60.png
[057d8a2ad7cf44dc879eba3da89e7721.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/ffa60ab37db944d38112bdc15beeccf7.png
[7d3922c77a5a4d75ac3f106a4ace1511.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/a49a4228c56644a8a3810217486ab31b.png
[a5cb94e4a4e24116ad25c6a195cffca1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/e94b20167b584966b5fdc39a31f4a5ce.png
[702ffeb4faa24dc2ade008a1d35d787a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/d4889eb7863f4d56981f3e0962f331aa.png
[4f7d039e78c6495f97cb0b6c58c69c72.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/6570c14184a2436fb39508e9dea3cdf7.png
[509326dbd2824378af5607ded7fc7017.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/da8d582f6ab748e493b55bc0403aea6b.jpeg