【RabbitMQ】之理论与实战

阳光穿透心脏的1/2处 2022-12-17 07:05 48阅读 0赞

# 一、MQ 简介 #

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

### 1、为什么使用MQ？MQ的优点 ###

简答：

*  异步处理 - 相比于传统的串行、并行方式，提高了系统吞吐量。
 *  应用解耦 - 系统间通过消息通信，不用关心其他系统的处理。
 *  流量削锋 - 可以通过消息队列长度控制请求量；可以缓解短时间内的高并发请求。
 *  日志处理 - 解决大量日志传输。
 *  消息通讯 - 消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通讯。比如实现点对点消息队列，或者聊天室等。

详答：

主要是：解耦、异步、削峰。

**解耦**：A 系统发送数据到 BCD 三个系统，通过接口调用发送。如果 E 系统也要这个数据呢？那如果 C 系统现在不需要了呢？A 系统负责人几乎崩溃…A 系统跟其它各种乱七八糟的系统严重耦合，A 系统产生一条比较关键的数据，很多系统都需要 A 系统将这个数据发送过来。如果使用 MQ，A 系统产生一条数据，发送到 MQ 里面去，哪个系统需要数据自己去 MQ 里面消费。如果新系统需要数据，直接从 MQ 里消费即可；如果某个系统不需要这条数据了，就取消对 MQ 消息的消费即可。这样下来，A 系统压根儿不需要去考虑要给谁发送数据，不需要维护这个代码，也不需要考虑人家是否调用成功、失败超时等情况。

就是一个系统或者一个模块，调用了多个系统或者模块，互相之间的调用很复杂，维护起来很麻烦。但是其实这个调用是不需要直接同步调用接口的，如果用 MQ 给它异步化解耦。

**异步**：A 系统接收一个请求，需要在自己本地写库，还需要在 BCD 三个系统写库，自己本地写库要 3ms，BCD 三个系统分别写库要 300ms、450ms、200ms。最终请求总延时是 3 + 300 + 450 + 200 = 953ms，接近 1s，用户感觉搞个什么东西，慢死了慢死了。用户通过浏览器发起请求。如果使用 MQ，那么 A 系统连续发送 3 条消息到 MQ 队列中，假如耗时 5ms，A 系统从接受一个请求到返回响应给用户，总时长是 3 + 5 = 8ms。

**削峰**：减少高峰时期对服务器压力。

### 2、MQ 的缺点 ###

**系统可用性降低**

本来系统运行好好的，现在你非要加入个消息队列进去，那消息队列挂了，你的系统不是呵呵了。因此，系统可用性会降低；

**系统复杂度提高**

加入了消息队列，要多考虑很多方面的问题，比如：一致性问题、如何保证消息不被重复消费、如何保证消息可靠性传输等。因此，需要考虑的东西更多，复杂性增大。

**一致性问题**

A 系统处理完了直接返回成功了，人都以为你这个请求就成功了；但是问题是，要是 BCD 三个系统那里，BD 两个系统写库成功了，结果 C 系统写库失败了，咋整？你这数据就不一致了。

所以消息队列实际是一种非常复杂的架构，你引入它有很多好处，但是也得针对它带来的坏处做各种额外的技术方案和架构来规避掉，做好之后，你会发现，妈呀，系统复杂度提升了一个数量级，也许是复杂了 10 倍。但是关键时刻，用，还是得用的。

### 3、MQ 的常见问题 ###

MQ 的常见问题有：

1.  消息的顺序问题
2.  消息的重复问题

**1）、消息的顺序问题**

消息有序指的是可以按照消息的发送顺序来消费。

假如生产者产生了 2 条消息：M1、M2，假定 M1 发送到 S1，M2 发送到 S2，如果要保证 M1 先于 M2 被消费，怎么做？

![img][]

解决方案：

（1）保证生产者 - MQServer - 消费者是一对一对一的关系

![img][img 1]

缺陷：

*  并行度就会成为消息系统的瓶颈（吞吐量不够）
 *  更多的异常处理，比如：只要消费端出现问题，就会导致整个处理流程阻塞，我们不得不花费更多的精力来解决阻塞的问题。

（2）通过合理的设计或者将问题分解来规避

*  不关注乱序的应用实际大量存在
 *  队列无序并不意味着消息无序，所以从业务层面来保证消息的顺序而不仅仅是依赖于消息系统，是一种更合理的方式。

**2）、消息的重复问题**

造成消息重复的根本原因是：**网络不可达**。所以解决这个问题的办法就是绕过这个问题。那么问题就变成了：如果消费端收到两条一样的消息，应该怎样处理？

解决的办法就是：消费端处理消息的业务逻辑应该保持幂等性。只要保持幂等性，不管来多少条重复消息，最后处理的结果都一样。

关于接口幂等性问题可以参考我的另一篇博客：[【编程开发】之接口的幂等性][Link 1]

### 4、MQ 相关概念 ###

**1）、消息服务中两个重要概念**

消息代理（message broker）和目的地（destination）；

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

**2）、消息队列主要的两种形式**

*  队列（queue）：点对点式
    
     *  –消息发送者发送消息，消息代理将其放入一个队列中，消息接收者从队列中获取消息内容，消息读取后被移出队列
     *  –消息只有唯一的发送者和接受者，但并不是说只能有一个接收者
 *  主题（topic）：发布订阅式
    
     *  –发送者（发布者）发送消息到主题，多个接收者（订阅者）监听（订阅）这个主题，那么就会在消息到达时同时收到消息

**3）、JMS（Java Message Service）JAVA消息服务**

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

JMS 是由 Java Api 定义，不跨语言也不跨平台，提供两种消息模型：点对点（Peer-2-Peer）和发布订阅（Pub/sub）。

支持多种消息类型：TextMessage、MapMessage、BytesMessage、StreamMessage、ObjectMessage、Message （只有消息头和属性）。

总结：JMS 定义了 JAVA API 层面的标准；在 java 体系中，多个 client 均可以通过JMS进行交互，不需要应用修改代码，但是其对跨平台的支持较差。

**4）、AMQP（Advanced Message Queuing Protocol）**

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

AMQP 由网络线级协议定义，即跨语言又跨平台，提供了五种消息模型：

*  （1）、direct exchange
 *  （2）、fanout exchange
 *  （3）、topic change
 *  （4）、headers exchange
 *  （5）、system exchange

本质来讲，后四种和JMS的pub/sub模型没有太大差别，仅是在路由机制上做了更详细的划分。

但消息类型只有一种：`byte[]`，当实际应用时，有复杂的消息，可以将消息序列化后发送。

总结：AMQP 定义了 wire-level 层的协议标准；天然具有跨平台、跨语言特性。

# 二、Rabbit MQ 简介 #

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

### 1、什么是 RabbitMQ ###

RabbitMQ 是一个由 erlang 开发的 AMQP(Advanved Message Queue Protocol) 的开源实现。

### 2、RabbitMQ 核心概念 ###

*  **Message**：消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。
 *  **Publisher**：消息的生产者，也是一个向交换器发布消息的客户端应用程序。
 *  **Exchange**：交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。Exchange有4种类型：direct(默认)，fanout，topic，和headers，不同类型的Exchange转发消息的策略有所区别
 *  **Queue**：消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。
 *  **Binding**：绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。Exchange 和 Queue 的绑定可以是多对多的关系。
 *  **Routing Key**： 路由关键字，exchange 根据这个关键字进行消息投递
 *  **Connection**：网络连接，比如一个TCP连接。
 *  **Channel**：信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
 *  **Consumer**：消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。
 *  **Virtual Host**：虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。
 *  **Broker**：表示消息队列服务器实体

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

### 3、RabbitMQ 运行机制 ###

#### AMQP 中的消息路由 ####

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

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

#### Exchange 类型 ####

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

**1）、Direct Exchange**

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

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

**2）、Fanout Exchange**

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

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

**3）、Topic Exchange**

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

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

### 4、RabbitMQ 的工作模式 ###

**1)、Simple 模式（最简单的的收发模式）**

*  消息产生消息，将消息放入队列
 *  消息的消费者监听消息队列，如果队列中有消息，就消费掉，消息被拿走后，自动从队列中删除

隐患：消息可能没有被消费者正确处理，已经从队列中消失了，造成消息的丢失，这里可以设置成手动的 ack，但如果设置成手动 ack，处理完后要及时发送 ack 消息给队列，否则会造成内存溢出。

**2）、Work 模式（资源的竞争）**

消息产生者将消息放入队列，多个消费者同时监听同一个队列，共同争抢当前的消息队列内容，谁先拿到谁负责消费消息。

隐患：高并发情况下，默认会产生某一个消息被多个消费者共同使用，可以设置一个开关(syncronize) 保证一条消息只能被一个消费者使用。

**3）、Publish/Subscribe 发布订阅模式（共享资源）**

*  每个消费者监听自己的队列；
 *  生产者将消息发给 broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息。

**4）、Routing 路由模式**

*  消息生产者将消息发送给交换机，交换机按照路由键匹配把消息发送给对应的消息队列；
 *  可以根据业务场景定义路由字符串。

**5）、Topic 主题模式（路由模式的一种）**

主题模式中的路由功能添加了模糊匹配：

*  `*` 号匹配一个单词
 *  `#` 号匹配 0 个或多个单词

工作流程如下：

*  消息产生者产生消息，把消息交给交换机
 *  交换机根据 key 的规则模糊匹配到对应的队列，由队列的监听消费者接收消息消费

### 5、解决 RabbitMQ 的消息问题 ###

**1）、消息的顺序性**

RabbitMQ 保证消息的顺序性的方法为：

*  1、拆分多个 queue，每个 queue 一个 consumer，这种方法会产生比较多的 queue，比较麻烦点；
 *  2、一个 queue 但是对应一个 consumer，然后这个 consumer 内部用内存队列做排队，然后分发给底层不同的 worker 来处理。

**2）、解决消息的丢失问题**

消息丢失分为：生产者丢失消息、消息列表丢失消息、消费者丢失消息；

*  生产者丢失消息：生产者开启确认模式，配置确认回调和回退回调
 *  消息列表丢失消息：开启消息持久化，创建队列的时候设置 durable 为 true，发送消息的时候设置 deliveryMode 为 2
 *  消费者丢失消息：设置为手动 Ack 应答模式

**3）、消息的重复消费问题**

保证消费消息的幂等性！

# 三、RabbitMQ 实战应用 #

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

### 1、SpringBoot 整合 RabbitMQ ###

**1）、引入依赖**

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

**2）、添加配置**

# 配置 RabbitMQ 服务器 IP 和端口
    spring.rabbitmq.host=192.168.56.10
    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

**3）、开启 RabbitMQ 功能**

*  1、引入了 amqp 场景启动器后，RabbitAutoConfiguration 就会自动生效
 *  2、SpringBoot 给容器自动配置了：RabbitTemplate、AmqpAdmin、CachingConnectionFactory、RabbitMessagingTemplate;
 *  3、在自动配置中，SpringBoot 使用的是 `@ConfigurationProperties(prefix = "spring.rabbitmq"`，所以，我们可以给配置文件中配置以 spring.rabbitmq 开头的 rabbitmq 相关信息；
 *  4、在主启动类中使用 `@EnableRabbit` 注解来开启 RabbitMQ 功能。

### 2、使用 RabbitMQ ###

要使用 RabbitMQ，我们需要创建交换机 Exchange、队列 Queue 和绑定 Binding 这三个对象，我们可以使用 AmqpAdmin 进行创建这三个对象。

**1）、创建 Exchange、Queue 、Binding**

@Autowired
    private AmqpAdmin amqpAdmin;
    
    // 创建 Exchange
    public void createExchange() { 
        // 创建一个 DirectExchange，参数为：交换机的名称、是否持久化、是否自动删除
        Exchange directExchange = new DirectExchange("hello-java-exchange", true, false);
        amqpAdmin.declareExchange(directExchange);
        log.info("Exchange[{}]创建成功：","hello-java-exchange");
    }
    
    // 创建 Queue
    public void testCreateQueue() { 
        // 队列参数为：队列名称、是否持久化、是否排他，是否自动删除
        Queue queue = new Queue("hello-java-queue", true, false, false);
        amqpAdmin.declareQueue(queue);
        log.info("Queue[{}]创建成功：","hello-java-queue");
    }
    
    // 创建 Binding
    public void createBinding() { 
        // destination【目的地】，如果【目的地类型】为队列则填写队列名称
        // destinationType【目的地类型】
        // exchange【交换机】
        // routingKey【路由键】
        // argument【自定义参数】
        // 功能：将交换机 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");
    
    }

**2）、发送消息**

发送消息使用的是 SpringBoot 给我们提供的 RabbitTemplate 来操作消息的发送：

@Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendMessageTest() { 
        // 发送字符串消息
        String msg = "Hello World";
        
        // 发送消息类型为自定义Java对象
        OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
        reasonEntity.setId(1L);
        reasonEntity.setCreateTime(new Date());
        reasonEntity.setName("reason");
        reasonEntity.setStatus(1);
        reasonEntity.setSort(2);   
    
        // 发送消息
        rabbitTemplate.convertAndSend("hello-java-exchange",     // 交换机
                                      "hello.java",              // 路由键 
                                      reasonEntity,				// 消息类型为对象
                                      new CorrelationData(UUID.randomUUID().toString()));
        log.info("消息发送完成:{}",reasonEntity);
    }

发送消息，既可以发送字符串消息也可以发送 Java 对象，如果发送的消息是个对象，会使用序列化机制，将对象写出去，对象必须实现 Serializable 接口。

如果想要把发送的对象转换为 JSON，我们可以自定消息转换器，通过自定义消息转换器可以把对象消息转换为 JSON。

@Configuration
    public class MyRabbitConfig { 
        // 自定义消息转换器
        @Bean
        public MessageConverter messageConverter() { 
            return new Jackson2JsonMessageConverter();
        }
    }

配置好后，再次调用 `rabbitTemplate.convertAndSend()` 方法时就会把消息对象转换成 JSON 对象。

**3）、接收消息**

RabbitMQ 接收消息使用的是 `@RabbitListener` 注解和 `@RabbitHandler` 注解，并且要使用 `@EnableRabbit` 注解开启 RabbitMQ 功能。

/** * @RabbitListener 属性： * queues：声明需要监听的所有队列 * * 接收消息函数参数类型： * 1、Message message：原生的消息详细消息：消息头 + 消息体 * 2、可以指定消息内容对应的对象，spring 帮我们自动转换：T<发送消息的类型> * 3、channel：当前传输数据的通道 */
    @RabbitListener(queues = { "hello-java-queue"})
    public void revieveMessage(Message message, 
                               OrderReturnReasonEntity content,
                               Channel channel) { 
        // 拿到主体内容
        byte[] body = message.getBody();
        // 拿到的消息头属性信息
        MessageProperties messageProperties = message.getMessageProperties();
        System.out.println("接受到的消息...内容" + message + "===内容：" + content);
    }

@RabbitListener 注解可以标注在类和方法上，而 @RabbitHandler 注解只能标注在方法上，所以我们一般使用 @RabbitListener 注解标注在类上表明监听的队列，而 @RabbitHandler 注解标注在方法上就可以进行方法并实现监听同一个队列的不同类型的消息：

@RabbitListener(queues = { "hello-java-queue"})
    public class OrderService { 
        
        @RabbitHandler 
        public void revieveMessage(Message message,  OrderReturnReasonEntity content) { 
        	System.out.println("接受到的消息...内容" + message + "===内容：" + content);
        }
        
        @RabbitHandler 
        public void revieveMessage2(Message message,  OrderEntity content) { 
        	System.out.println("接受到的消息...内容" + message + "===内容：" + content);
        }
        
    }

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

由于网络的卡顿原因，或者服务宕机了都可能会导致的消息丢失，消息的丢失包括生产者丢失消息、消息列表丢失消息、消费者丢失消息。为了保证我们发送的消息不丢失、可靠抵达我们以前可以使用事务消息来解决，但这种方式性能下降严重，所以就引入了消息确认机制。

由于生产者发送的消息涉及两步：

*  1、生产者发送消息给 Broker；
 *  2、Broker 收到消息后把消息交给 Exchange，并由 Exchange 根据路由键交给指定的 Queue。

所以生产者端的确认机制涉及两种：

*  1、生产者到 Broker 的 ConfirmCallback 确认模式；
 *  2、Exchange 到 Queue 到 ReturnCallback 回退模式。

而消费者使用的是 Ack 应答机制。

**1）、生产者确认机制**

为了实现生产者的确认机制，我需要自定义 RabbitTemplate ：

@Configuration
    public class MyRabbitConfig { 
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
        
        @PostConstruct  // MyRabbitConfig 对象创建完成以后，执行这个方法
        public void initRabbitTemplate() { 
    
            /** * 设置确认回调，只要消息抵达Broker就ack=true * correlationData：当前消息的唯一关联数据(这个是消息的唯一id) * ack：消息是否成功收到 * cause：失败的原因 */
            rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { 
                System.out.println("确认回调：" + correlationData + ack + cause);
            });
    
    
            /** * 设置 Return 回调，只要消息没有投递给指定的队列，就触发这个失败回调 * message：投递失败的消息详细信息 * replyCode：回复的状态码 * replyText：回复的文本内容 * exchange：当时这个消息发给哪个交换机 * routingKey：当时这个消息用哪个路邮键 */
            rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey) -> { 
                System.out.println("Return 回调：" + message + replyCode + replyText + exchange + routingKey);
            });
        }
    }

并且需要在配置文件中开启相关设置：

# 开启发送端消息抵达Broker确认
    spring.rabbitmq.publisher-confirms=true
    # 开启发送端消息抵达Queue确认
    spring.rabbitmq.publisher-returns=true
    # 只要消息抵达Queue，就会异步发送优先回调returnfirm
    spring.rabbitmq.template.mandatory=true

**2）、消费者确认机制**

RabbitMQ 消费者默认是自动确认的，只要消息收到，客户端会自动确认，所以，不管你消息有没有处理成功都会把消息从消息队列中删除。

所以为了消费者的消息可靠，我们需要修改消息为手动确认，修改为手动确认需要修改配置文件内容：

# 手动ack消息，不使用默认的消费端确认
    spring.rabbitmq.listener.simple.acknowledge-mode=manual

设置为手动确认模式后，只要我们没有明确告诉 MQ 消息被签收，即没有 Ack 时，消息就是一直 unacked 状态，即使 Consumer 宕机了，消息也不会丢失，会重新变为 Ready 状态，下一次有新 客户端连接时再返回给客户端。

消费者消息签收确认代码如下：

@RabbitListener(queues = { "hello-java-queue"})
    public void revieveMessage(Message message, 
                               OrderReturnReasonEntity content,
                               Channel channel) { 
        System.out.println("受到的消息..." + content);
        
        // deliveryTag：消息投递标签，为一个 channel 内按顺序自增数字
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        
        try { 
            // 处理成功则签收：1、签收的 deliveryTag；2、是否批量签收
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) { 
            // 处理失败退签：1、退签的 deliveryTag；2、是否批量操作；3、是否把消息返回给服务器
            channel.basicNack(deliveryTag, false, true)
        }
    }

# 四、RabbitMQ 延时队列（实现定时任务） #

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

### 1、定时任务的场景 ###

购物下单等待支付的订单，超过一定时间后，系统就会自动取消订单并释放占有的资源。

常用的解决方案是：使用 Spring 的 Schedule 定时任务轮询数据库。但缺点很大：消耗系统内存、增加了数据库的压力、存在较大的时间误差。

较好的解决方案是使用 RabbitMQ 的延时队列：RabbitMQ 消息的 TTL 和死信 Exchange 结合。

### 2、消息的 TTL（Time To Live） ###

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

*  对队列设置 TTL 就是队列没有消费者连着的保留时间，也可以对每个消息单独设置。超过了这个时间，我们就认为这个消息死了，这种消息称之为**死信**，而保存死信的队列就被称之为**死信队列**。
 *  如果同时对队列和消息设置，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息的死亡时间有可能不一样。

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

### 3、死信路由交换机 - Dead Letter Exchange（DLX） ###

一个消息在满足如下条件时会被放入到**死信路由**中（什么是死信）：

*  一个消息被 Consumer 拒收了，并且 reject 方法的参数里 requque 是 false，也就是说不会被再次放入到队列中被其它消费者消费；
 *  消息的 TTL 到了，消息过期了；
 *  队列的长度限制满了，排在前面的消息会被丢弃或者扔到私信路由上。

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

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

### 4、延时队列原理 ###

![在这里插入图片描述][20201024220955254.PNG]

其工作原理和流程如下：

1.  延时队列 `order-delay-queue` 和消费者监听的队列 `order-release-queue` 同时和 `order-event-exchange` 交换机绑定，延时队列和交换机的路由键为 `order-create`，消费者队列的路由键为 `order-release-queue`；
2.  生产者发送消息到 exchange 中，路由键设置为 `order-create`，然后交换机把消息投递到延时队列中；
3.  由于消息设置了超时时间 `x-message-ttl` 为 60 秒，死信队列路由交换机为 `order-event-exchange`，死信路由键为 `order-release`，所以当存放的消息到达指定的 60 秒后，就会变成私信并投递到指定的死信路由交换机中，这样，消息又重新回到了的了 `order-event-exchange` 交换机中；
4.  由于死信路由键已经重新设置为了 `order-release`，所以，交换机再次根据该路由把消息投递到 `order-release-queue` 队列中，最后交给消费者处理。

在整个流程中，我们服用了 `order-event-exchange` 交换机，而延时队列只作为存放延时消息的队列不被任何消费者监听。

### 5、延时队列应用案例 ###

根据上面的原理和流程，下面我们来编写一个延时队列的案例：

import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.Exchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    
    @Configuration
    public class MyRabbitMQConfig { 
    
        // 把 Queue、Exchange、Binding 对象的创建交给 Spring 管理
    
        // 创建交换机
        @Bean
        public Exchange orderEventExchange() { 
            /* 交换机类型为 TopicExchange * String name, * boolean durable, * boolean autoDelete, * Map<String, Object> arguments */
            return new TopicExchange("order-event-exchange", true, false);
    
        }
    
        // 创建死信队列
        @Bean
        public Queue orderDelayQueue() { 
            /* * Queue(String name, 队列名字 * boolean durable, 是否持久化 * boolean exclusive, 是否排他 * boolean autoDelete, 是否自动删除 * Map<String, Object> arguments) 属性 */
            HashMap<String, Object> arguments = new HashMap<>();
            arguments.put("x-dead-letter-exchange", "order-event-exchange");
            arguments.put("x-dead-letter-routing-key", "order.release");
            arguments.put("x-message-ttl", 60000); // 消息过期时间 1分钟
            Queue queue = new Queue("order.delay.queue", true, false, false, arguments);
    
            return queue;
        }
    
        // 消费者队列
        @Bean
        public Queue orderReleaseQueue() { 
    
            Queue queue = new Queue("order.release.queue", true, false, false);
    
            return queue;
        }
    
        // 绑定延时队列和交换机
        @Bean
        public Binding orderCreateBinding() { 
            /* * String destination, 目的地（队列名或者交换机名字） * DestinationType destinationType, 目的地类型（Queue、Exhcange） * String exchange, * String routingKey, * Map<String, Object> arguments * */
            return new Binding("order.delay.queue", Binding.DestinationType.QUEUE,
                               "order-event-exchange", "order.create", null);
        }
    
        // 绑定消费者队列和交换机
        @Bean
        public Binding orderReleaseBinding() { 
    
            return new Binding("order.release.queue", Binding.DestinationType.QUEUE,
                               "order-event-exchange", "order.release.order", null);
        }
    
    }

[img]: /images/20221123/f9c6a98c58bf467d90e10b81c01ca80c.png
[img 1]: /images/20221123/c266822aaffc45998de9e3ee0b8ec120.png
[Link 1]: https://blog.csdn.net/aiwangtingyun/article/details/109233690
[20201022114047930.png]: /images/20221123/5922cc9e2c604dee8774af6bba6457ac.png
[20201022114107945.png]: /images/20221123/dae3b44d42cb465ea2d54ecb710f0b4c.png
[20201022114122173.png]: /images/20221123/5ffaf365ce1945adb9e9c7a021c3ba16.png
[2020102211413561.png]: /images/20221123/518c7b828d0e4f5d962dd4e0b1601de8.png
[20201022114153726.png]: /images/20221123/fbe7b0b2348548f3874b63fcc87260c8.png
[20201024220955254.PNG]: /images/20221123/c7d804fc51b948e797dcf5cb4ffe5d9d.png