【RabbitMQ】掌握高级特性，为你的应用带来无限可能！

悠悠 2024-03-26 22:54 76阅读 0赞

#### RabbitMQ高级特性 ####

*  1. 什么是MQ？
 *   *  1.1 应用解耦
     *  1.2 消息流量削峰
     *  1.3 数据分发
 *  2. 各种MQ产品比较
 *  3. 消息可靠性
 *   *  3.1 生产者消息确认
     *  3.2 消息持久化
     *  3.3 消费者消息确认
     *  3.4 消费失败重试机制
 *  4. 死信交换机
 *  5. TTL
 *  6. 延迟队列
 *   *  6.1 安装DelayExchange插件
     *  6.2 延迟队列的使用
 *  7. 惰性队列

## 1. 什么是MQ？ ##

MQ指的是消息队列，而消息队列是一种应用程序之间进行异步通信的机制，常用于分布式系统中传递消息和实现解耦。

![ef047e3060bb7a593ef7358e047d54ba.png][]

同时，MQ也是一种先进先出的数据结构，消息会按照进入MQ的顺序依次被消费者消费。

在开发中，MQ的应用应用场景包含但不限于下面的几种。

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

### 1.1 应用解耦 ###

应用解耦是减少是指**减少程序中每个功能之间的依赖关系**。

就好比一个订单系统（简化版），一个订单的完成需要有以下四部分

1.  订单信息插入数据库
2.  支付系统扣减金额
3.  库存系统扣减库存
4.  物流系统生成物流信息

很明显，这里涉及了四个系统，假如这些功能的实现是采用类似Dubbo这样的RPC框架实现的话，若第2、3、4部分任意一个系统出现异常或者升级都会导致下单功能不可用，这在一定情况下是非常致命的，非常影响用户体验。

![外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传][64aeb072240a4eefbbb1ecb1df5391a5.png]

这时候，我们可以引入MQ，将这些系统解耦，订单成功插入数据库后，将订单信息放到MQ中发送给其他三个系统，其他三个系统拉取MQ的消息，对消息进行消费，这个过程中，就算其他三个系统出现了异常，只需要恢复后重新消费MQ里面的信息即可，不会影响到用户的下单体验。

这样我们就完成了对这个订单系统的解耦，系统的耦合性就会降低了，容错性也会提高。

![外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传][d9a09ae8fbb4416096e306b0a1637a67.jpeg]

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

### 1.2 消息流量削峰 ###

在没有MQ的系统里面，系统架构大致上是这样的，用户的请求直接打到系统中，系统会根据请求做出相应处理随后与MySQL进行交互。![c020e9948466a82b8bf8a9fc57a88c93.png][]

这样看，这个架构是没有什么问题的，但是应用系统如果遇到系统请求流量的瞬间猛增，比如双十一这样的秒杀场景下，可能会用同时百万个用户请求直接达到服务端。对于服务端来说，也许可能能扛下这个百万请求，但是对于MySQL来说，这百万请求无疑是压死骆驼的最后一根稻草，会导致数据库挂掉。

这时候，引入MQ做消息流量削峰可以有效降低MySQL的压力。

消息流量削峰顾名思义就是将**请求流量的峰值压低**，具体是如何操作的呢？

比如用户每秒五千个请求，这些请求可以先使用MQ缓存起来，然后A系统每秒从MQ中拉取两千个请求，等待A系统处理完毕后，再拉取新的数据处理，这样虽然用户体验起来会感觉速度明显下降，但是总比MySQL挂了强。

![defa292d692ecfc1ca129e22991c860f.png][]

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

### 1.3 数据分发 ###

对于A系统来说，A系统处理完成的数据，可能需要被B、C、D等多个系统使用，这时候如果在同一个接口使用RPC调用B、C、D等多个系统的服务会导致A系统可维护性非常差，万一那天B系统不需要了，就需要修改A系统的代码，这样维护起来成本太高。  
![0cfc48bd44c085f5b66ac1095294138d.png][]

其实A系统根本不需要在乎谁需要我处理完成的数据，只需要将数据发送给MQ，数据的使用方去MQ拉取需要的数据即可，这样即使其他系统以后变动不再需要A处理的数据，也不用更改A程序的代码。

![13fddcf60c6689db206ce45ab31ab61d.png][]

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

## 2. 各种MQ产品比较 ##

现如今，企业比较常用的MQ主要有：

1.  ActiveMQ
2.  RabbitMQ
3.  RocketMQ
4.  Kafka

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th><strong>RabbitMQ</strong></th> 
   <th><strong>ActiveMQ</strong></th> 
   <th><strong>RocketMQ</strong></th> 
   <th><strong>Kafka</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>公司/社区</td> 
   <td>Rabbit</td> 
   <td>Apache</td> 
   <td>阿里</td> 
   <td>Apache</td> 
  </tr> 
  <tr> 
   <td>开发语言</td> 
   <td>Erlang</td> 
   <td>Java</td> 
   <td>Java</td> 
   <td>Scala&amp;Java</td> 
  </tr> 
  <tr> 
   <td>协议支持</td> 
   <td>AMQP，XMPP，SMTP，STOMP</td> 
   <td>OpenWire,STOMP，REST,XMPP,AMQP</td> 
   <td>自定义协议</td> 
   <td>自定义协议</td> 
  </tr> 
  <tr> 
   <td>可用性</td> 
   <td>高</td> 
   <td>一般</td> 
   <td>高</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>单机吞吐量</td> 
   <td>一般</td> 
   <td>差</td> 
   <td>高</td> 
   <td>非常高</td> 
  </tr> 
  <tr> 
   <td>消息延迟</td> 
   <td>微秒级</td> 
   <td>毫秒级</td> 
   <td>毫秒级</td> 
   <td>毫秒以内</td> 
  </tr> 
  <tr> 
   <td>消息可靠性</td> 
   <td>高</td> 
   <td>一般</td> 
   <td>高</td> 
   <td>一般</td> 
  </tr> 
  <tr> 
   <td>功能特性</td> 
   <td>基于erlang开发，所以并发能力很强，性能极其好，延时很低；管理界面丰富</td> 
   <td>成熟产品，在很多公司得到应用；有较多的文档；各种协议支持较好</td> 
   <td>MQ功能比较完备，扩展性佳</td> 
   <td>只支持主要的MQ功能，像一些消息查询，消息回溯等功能没有提供，毕竟是为大数据准备的，在大数据领域应用广</td> 
  </tr> 
 </tbody> 
</table>

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

## 3. 消息可靠性 ##

前面简单介绍了一下MQ的作用了比较，接下来就得谈谈RabbitMQ的进阶了。

消息可靠性是指消息从发送到消费者接收，这一过程的可靠性。

这一过程是指消息生产者把消息发送至交换机，交换机将消息发送到队列，消费者从队列里面获取消息，这一连串的过程。

![image-20210718155059371][]

***这其中每一步导致丢失的原因都不一样：***

1.  发送时候丢失
    
     *  生产者发送的消息未送达exchange
     *  消息到达exchange后未到达queue
2.  MQ宕机，queue将消息丢失
3.  消费者接收到消息后未消费就宕机

***针对与这些情况，RabbitMQ给出了相应的解决方案：***

1.  生产者确认机制
2.  MQ持久化
3.  消费者确认机制
4.  失败重试机制

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

### 3.1 生产者消息确认 ###

RabbitMQ提供了生产者消息确认的机制，**这个机制必须要给每个消息指定一个唯一的ID，目的是为了区别不同的消息，防止ACK冲突**。消息发送到MQ以后，会返回一个结果给发送者，表示消息是否处理成功。如果没有处理功能，可以让MQ重发消息。

返回的结果有两种方式：

1.  **`publisher-confirm`**，发送者确认
    
     *  消息成功投递到交换机，返回ack
     *  消息未投递到交换机，返回nack
2.  **`publisher-return`**，发送者回执
    
     *  消息投递到交换机了，但是没有路由到队列。返回ACK，以及路由失败原因。
    
    ![image-20210718160907166][]

要想实现这一功能，首先需要修改生产者服务的`application.yml`文件，添加下面的内容：

spring:
      rabbitmq:
        publisher-confirm-type: correlated
        publisher-returns: true
        template:
          mandatory: true

说明：

*  `publish-confirm-type`：开启publisher-confirm，这里支持两种类型：
    
     *  `simple`：同步等待confirm结果，直到超时
     *  `correlated`：异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback
 *  `publish-returns`：开启publish-return功能，同样是基于callback机制，不过是定义ReturnCallback
 *  `template.mandatory`：定义消息路由失败时的策略。true，则调用ReturnCallback；false：则直接丢弃消息

接着，**需要定义Return回调**，因为每个`RabbitTemplate`只能配置一个`ReturnCallback`，所以可以利用Spring的`ApplicationContextAware`实现项目加载配置的时候就配置完成。

@Slf4j
    @Configuration
    public class CommonConfig implements ApplicationContextAware {
        
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        
            // 获取RabbitTemplate
            RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
            // 设置ReturnCallback
            rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
        
                // 投递失败，记录日志
                log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息{}",
                         replyCode, replyText, exchange, routingKey, message.toString());
                // 如果有业务需要，可以重发消息
            });
        }
    }

然后，定义`ConfirmCallback`，`ConfirmCallback`可以在发送消息的时候指定，因为每个业务confirm成功或失败的逻辑不一定相同。

public void testSendMessage2SimpleQueue() throws InterruptedException {
        
        // 1.消息体
        String message = "hello, spring amqp!";
        // 2.全局唯一的消息ID，需要封装到CorrelationData中
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        // 3.添加callback
        correlationData.getFuture().addCallback(
            result -> {
        
                if(result.isAck()){
        
                    // 3.1.ack，消息成功
                    log.debug("消息发送成功, ID:{}", correlationData.getId());
                }else{
        
                    // 3.2.nack，消息失败
                    log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
                }
            },
            ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
        );
        // 4.发送消息
        rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);
    
        // 休眠一会儿，等待ack回执
        Thread.sleep(2000);
    }

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

### 3.2 消息持久化 ###

生产者消息确认机制可以很好地保证消息被投放到RabbitMQ的队列中，但是如果RabbitMQ突然宕机，可能会导致消息丢失，而消息持久化，就是为了解决这一情况的出现。

***消息持久化分为三种：***

1.  交换机持久化
2.  队列持久化
3.  消息持久化

***第一种，交换机持久化，这种实现起来比较容易，只需要在定义交换机的时候设置成持久化即可***

@Bean
    public DirectExchange simpleExchange(){
        
        // 三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除
        return new DirectExchange("simple.direct", true, false);
    }

默认情况下，由`SpringAMQP`声明的交换机都是持久化的。

持久化的交换机在图形化控制台看都是带一个`D`的

***第二种，队列持久化，这种也是在定义队列的时候设置成持久化即可***

@Bean
    public Queue simpleQueue(){
        // 使用QueueBuilder构建队列，durable就是持久化的
        return QueueBuilder.durable("simple.queue").build();
    }

默认情况下，由`SpringAMQP`声明的队列都是持久化的。

持久化的交换机在图形化控制台看也都是带一个`D`的

***第三种，消息持久化，这种在发送消息的时候，可以设置消息的属性，可以指定`delivery-mode`为持久化还是非持久化***

@Test
    public void testSendDelayMessage() throws InterruptedException {
        // 1.准备消息
        Message message = MessageBuilder
            .withBody("hello, ttl messsage".getBytes(StandardCharsets.UTF_8))
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
            .setHeader("x-delay", 5000)
            .build();
        // 2.准备CorrelationData
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        // 3.发送消息
        rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);
    
        log.info("发送消息成功");
    }

默认情况下，`SpringAMQP`发出的任何消息都是持久化的，不用特意指定。

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

### 3.3 消费者消息确认 ###

RabbitMQ**确认消息被消费者消费后会立马删除**，而RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的：消费者获取消息后，应该向RabbitMQ发送ACK回执，表明自己已经处理消息。

***`SpringAMQP`则允许配置三种确认模式：***

•`manual`：手动ack，需要在业务代码结束后，调用api发送ack。

•`auto`：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack

•`none`：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除

由此可知：

*  none模式下，消息投递是不可靠的，可能丢失
 *  auto模式类似事务机制，出现异常时返回nack，消息回滚到mq；没有异常，返回ack
 *  manual：自己根据业务情况，判断什么时候该ack

**一般，我们都是使用默认的auto即可。**

那么该如何使用auto模式，只需要在`yml`配置文件将确认机制改为auto即可。

spring:
      rabbitmq:
        listener:
          simple:
            acknowledge-mode: auto # 关闭ack

当出现异常的时候，在控制台可能看到消息为unack(未确认状态)

![image-20210718171705383][]

抛出异常后，因为Spring会自动返回nack，所以消息恢复至Ready状态，并且没有被RabbitMQ删除：

![image-20210718171759179][]

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

### 3.4 消费失败重试机制 ###

当消费者出现异常后，消息会不断requeue（重入队）到队列，再重新发送给消费者，然后再次异常，再次requeue，无限循环，导致mq的消息处理飙升，带来不必要的压力

![image-20210718172746378][]

***解决方案无疑就是本地重试***

***本地重试，是指可以利用Spring的retry机制，在消费者出现异常时利用本地重试，而不是无限制的requeue到MQ队列***

需要修改消费者服务的`application.yml`文件，添加内容：

spring:
      rabbitmq:
        listener:
          simple:
            retry:
              enabled: true # 开启消费者失败重试
              initial-interval: 1000 # 初识的失败等待时长为1秒
              multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
              max-attempts: 3 # 最大重试次数
              stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

这种方案重试达到最大次数后，Spring会返回ack，消息会被丢弃。

但在本地重试达到最大重试次数后，消息会被丢弃，这是由Spring内部机制决定的。

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

*  `RejectAndDontRequeueRecoverer`：重试耗尽后，直接reject，丢弃消息。默认就是这种方式
 *  `ImmediateRequeueMessageRecoverer`：重试耗尽后，返回nack，消息重新入队
 *  `RepublishMessageRecoverer`：重试耗尽后，将失败消息投递到指定的交换机

比较优雅的一种处理方案是`RepublishMessageRecoverer`，失败后将消息投递到一个指定的，**专门存放异常消息的队列，后续由人工集中处理。**

这时候，**需要在消费者服务中定义失败的消息的交换机和队列**

@Bean
    public DirectExchange errorMessageExchange(){
        return new DirectExchange("error.direct");
    }
    @Bean
    public Queue errorQueue(){
        return new Queue("error.queue", true);
    }
    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
        return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
    }

接着定义一个`RepublishMessageRecoverer`，关联队列和交换机

@Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }

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

## 4. 死信交换机 ##

当一个队列中的消息满足下列情况之一时，可以成为**死信（dead letter）**：

*  消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
 *  消息是一个过期消息，超时无人消费
 *  要投递的队列消息满了，无法投递

如果这个包含死信的队列配置了`dead-letter-exchange`属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为**死信交换机**（Dead Letter Exchange，检查DLX）。

如图，一个消息被消费者拒绝了，变成了死信：

![image-20210718174328383][]

因为simple.queue绑定了死信交换机 dl.direct，因此死信会投递给这个交换机：

![image-20210718174416160][]

如果这个死信交换机也绑定了一个队列，则消息最终会进入这个存放死信的队列：

![image-20210718174506856][]

另外，队列将死信投递给死信交换机时，必须知道两个信息：

*  死信交换机名称
 *  死信交换机与死信队列绑定的RoutingKey

这样才能确保投递的消息能到达死信交换机，并且正确的路由到死信队列。

![image-20210821073801398][]

**在失败重试策略中，默认的`RejectAndDontRequeueRecoverer`会在本地重试次数耗尽后，发送reject给RabbitMQ，消息变成死信，被丢弃。**

**我们可以给`simple.queue`添加一个死信交换机，给死信交换机绑定一个队列。这样消息变成死信后也不会丢弃，而是最终投递到死信交换机，路由到与死信交换机绑定的队列。**

![image-20210718174506856][]

在消费者服务中，定义一组死信交换机、死信队列，这样死信就会都在死心队列中了

// 声明普通的 simple.queue队列，并且为其指定死信交换机：dl.direct
    @Bean
    public Queue simpleQueue2(){
        
        return QueueBuilder.durable("simple.queue") // 指定队列名称，并持久化
            .deadLetterExchange("dl.direct") // 指定死信交换机
            .build();
    }
    // 声明死信交换机 dl.direct
    @Bean
    public DirectExchange dlExchange(){
        
        return new DirectExchange("dl.direct", true, false);
    }
    // 声明存储死信的队列 dl.queue
    @Bean
    public Queue dlQueue(){
        
        return new Queue("dl.queue", true);
    }
    // 将死信队列 与 死信交换机绑定
    @Bean
    public Binding dlBinding(){
        
        return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");
    }

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

## 5. TTL ##

**一个队列中的消息如果超时未消费**，则会变为死信，超时分为两种情况：

*  消息所在的队列设置了超时时间
 *  消息本身设置了超时时间

利用这个特性，就可以做到延迟消息的发送了，比如用于订单十五分钟后取消，这样的操作。

![image-20210718182643311][]

在消费者服务的`SpringRabbitListener`中定义一个新的消费者，声明死信交换机、死信队列

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "dl.ttl.queue", durable = "true"),
        exchange = @Exchange(name = "dl.ttl.direct"),
        key = "ttl"
    ))
    public void listenDlQueue(String msg){
        log.info("接收到 dl.ttl.queue的延迟消息：{}", msg);
    }

声明一个队列，并指定TTL

要给队列设置超时时间，需要在声明队列时配置`x-message-ttl`属性：

@Bean
    public Queue ttlQueue(){
        return QueueBuilder.durable("ttl.queue") // 指定队列名称，并持久化
            .ttl(10000) // 设置队列的超时时间，10秒
            .deadLetterExchange("dl.ttl.direct") // 指定死信交换机
            .build();
    }

注意，这个队列设定了死信交换机为`dl.ttl.direct`

接着声明交换机，将ttl与交换机绑定

@Bean
    public DirectExchange ttlExchange(){
        return new DirectExchange("ttl.direct");
    }
    @Bean
    public Binding ttlBinding(){
        return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
    }

接着，发送信息，但不要指定TTL

@Test
    public void testTTLQueue() {
        // 创建消息
        String message = "hello, ttl queue";
        // 消息ID，需要封装到CorrelationData中
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        // 发送消息
        rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
        // 记录日志
        log.debug("发送消息成功");
    }

发送消息的日志：

![image-20210718191657478][]

查看下接收消息的日志：

![image-20210718191738706][]

因为队列的TTL值是10000ms，也就是10秒。可以看到消息发送与接收之间的时差刚好是10秒。

除此之外，还能在发送消息的时候，指定TTL

@Test
    public void testTTLMsg() {
        // 创建消息
        Message message = MessageBuilder
            .withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8))
            .setExpiration("5000")
            .build();
        // 消息ID，需要封装到CorrelationData中
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        // 发送消息
        rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
        log.debug("发送消息成功");
    }

查看发送消息日志：

![image-20210718191939140][]

接收消息日志：

![image-20210718192004662][]

这次，发送与接收的延迟只有5秒。说明当队列、消息都设置了TTL时，任意一个到期就会成为死信。

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

## 6. 延迟队列 ##

上面就是原生实现的一种延迟消息队列，但是因为延迟队列的应用场景很多，RabbitMQ也推出了插件，原生支持延迟队列的效果。

这个插件就是DelayExchange插件。参考RabbitMQ的插件列表页面：https://www.rabbitmq.com/community-plugins.html

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

### 6.1 安装DelayExchange插件 ###

官方的安装指南地址为：https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq

上述文档是基于linux原生安装RabbitMQ，然后安装插件。

RabbitMQ有一个官方的插件社区，地址为：https://www.rabbitmq.com/community-plugins.html

其中包含各种各样的插件，包括我们要使用的DelayExchange插件：

![image-20210713104511055][]

大家可以去对应的GitHub页面下载3.8.9版本的插件，地址为https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/tag/3.8.9这个对应RabbitMQ的3.8.5以上版本。

因为我们是基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。如果不是基于Docker的同学，请参考第一章部分，重新创建Docker容器。

我们之前设定的RabbitMQ的数据卷名称为`mq-plugins`，所以我们使用下面命令查看数据卷：

docker volume inspect mq-plugins

可以得到下面结果：

![image-20210713105135701][]

接下来，将插件上传到这个目录即可：

![image-20210713105339785][]

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

### 6.2 延迟队列的使用 ###

DelayExchange需要将一个交换机声明为delayed类型。当我们发送消息到delayExchange时，流程如下：

*  接收消息
 *  判断消息是否具备x-delay属性
 *  如果有x-delay属性，说明是延迟消息，持久化到硬盘，读取x-delay值，作为延迟时间
 *  返回routing not found结果给消息发送者
 *  x-delay时间到期后，重新投递消息到指定队列

插件的使用也非常简单：声明一个交换机，交换机的类型可以是任意类型，只需要设定delayed属性为true即可，然后声明队列与其绑定即可。

首先，声明一个`DelayExchange`交换机

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "delay.queue", durable = "true"),
        exchange = @Exchange(name = "delay.direct", delayed = "true"),
        key = "delay"
    ))
    public void listenDelayExchange(String msg) {
        log.info("消费者接收到了delay.queue的延迟消息");
    }

也可以通过`@Bean`的方式

@Bean
    public DirectExchange directExchange(){
        return  ExchangeBuilder
            // 指定交换机的名称和类型
            .directExchange("delay.direct")
            //设置delay属性为true
            .delayed()
            //持久化
            .durable(true)
            .build();
    
    }

接着，发送消息

@Test
        public void testSendDelayMessage() throws InterruptedException {
            // 1.准备消息
            Message message = MessageBuilder
                    .withBody("hello, ttl messsage".getBytes(StandardCharsets.UTF_8))
                    .setHeader("x-delay", 5000)
                    .build();
            // 2.准备CorrelationData，封装到CorrelationData里面
            CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
            // 3.发送消息
            rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);
    
            log.info("发送消息成功");
        }

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

## 7. 惰性队列 ##

当生产者发送消息的速度超过了消费者处理消息的速度，就会导致**队列中的消息堆积**，直到队列存储消息达到上限。**之后发送的消息就会成为死信，可能会被丢弃，这就是消息堆积问题。**

![image-20210718194040498][]

***解决消息堆积有两种思路：***

*  增加更多消费者，提高消费速度。也就是我们之前说的`work queue`模式
 *  扩大队列容积，提高堆积上限

从RabbitMQ的3.6.0版本开始，就增加了`Lazy Queues`的概念，也就是惰性队列。**惰性队列的特征如下**：

*  接收到消息后直接存入磁盘而非内存
 *  消费者要消费消息时才会从磁盘中读取并加载到内存
 *  支持数百万条的消息存储

***要设置一个队列为惰性队列，只需要在声明队列时，指定x-queue-mode属性为lazy即可。可以通过命令行将一个运行中的队列修改为惰性队列：***

rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues

命令解读：

*  `rabbitmqctl` ：RabbitMQ的命令行工具
 *  `set_policy` ：添加一个策略
 *  `Lazy` ：策略名称，可以自定义
 *  `"^lazy-queue$"` ：用正则表达式匹配队列的名字
 *  `'{"queue-mode":"lazy"}'` ：设置队列模式为lazy模式
 *  `--apply-to queues`：策略的作用对象，是所有的队列

也可以通过`@Bean`声明`lazy-queue`

@Bean
    public Queue lazyQueue() {
        return QueueBuilder.durable("lazy.queue")
            // 开启x-queue-mode为lazy
            .lazy()
            .build();
    }

也可以基于@RabbitListener声明LazyQueue

@RabbitListener(queuesToDeclare = @Queue(
        name = "lazy.queue",
        durable = "true",
        arguments = @Argument(name = "x-queue-mode", value = "lazy")
    ))
    public void listenLazyQueue(String msg){
        log.info("收到lazy.queue的消息：{}", msg);
    }

虽然惰性队列能解决消息堆积的问题，但是也存在一下缺点

*  基于磁盘存储，消息时效性会降低
 *  性能受限于磁盘的IO

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

> **参考：**
> 
> 1.  [SpringCloud+RabbitMQ+Docker+Redis+搜索+分布式，系统详解springcloud微服务技术栈课程][SpringCloud_RabbitMQ_Docker_Redis_springcloud]

[ef047e3060bb7a593ef7358e047d54ba.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/806d531abbe240e8a0b463206075a477.png
[64aeb072240a4eefbbb1ecb1df5391a5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/da8464dee5454f11b4a98e68db1e2dd8.png
[d9a09ae8fbb4416096e306b0a1637a67.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/6849cb143f764a44a8d6b242aa73ffb0.png
[c020e9948466a82b8bf8a9fc57a88c93.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/70b92204147a4df2a3b9d54b686c14d4.png
[defa292d692ecfc1ca129e22991c860f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/589b2d8f1a5b4d30bbe2866565c303e2.png
[0cfc48bd44c085f5b66ac1095294138d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/35c3248536f1416789f1a13b7061cbcb.png
[13fddcf60c6689db206ce45ab31ab61d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/4ebcba53f64344d8b3c380ad310053ba.png
[image-20210718155059371]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/9b5c2f6c296744d08230db6d6bd91623.png
[image-20210718160907166]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/43cf07259bc74eee817c60a44a72fcd7.png
[image-20210718171705383]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/4a8e17c7ce0842eb86ab222e49a8b37c.png
[image-20210718171759179]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/7c6a7a19ba1441f988f6d9148ca86e0a.png
[image-20210718172746378]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/6f8500fe0e1f49b1bad488469c54a853.png
[image-20210718174328383]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/cd593830009e46179b8509b57afd8374.png
[image-20210718174416160]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/7ec984bfb492465e9042e3d15e916236.png
[image-20210718174506856]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/52033eddca0444e48a19cda90ff77f4e.png
[image-20210821073801398]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/f69908ed395340a3a05dbcbc05fea6ae.png
[image-20210718182643311]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/31dc6427834c4889946b3c22409144a0.png
[image-20210718191657478]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/8f6806f4e7a4493389983dd54ea4cdde.png
[image-20210718191738706]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/b93bee84ac534a51942484f6f032eca9.png
[image-20210718191939140]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/b2c8cd77eb074ea39999d5977b1e9d9d.png
[image-20210718192004662]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/dbc18e3f5d0249e1b90cc3ab1071e1bc.png
[image-20210713104511055]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/41fc7e8cc5764657ba3f2bcf531751c0.png
[image-20210713105135701]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/40641bd750a24bd9982baa6d3f5507a8.png
[image-20210713105339785]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/b41d2efcc4774f3184fd4897da694f7d.png
[image-20210718194040498]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/d7f86953d0894a43a4074cf1e66898d0.png
[SpringCloud_RabbitMQ_Docker_Redis_springcloud]: https://www.bilibili.com/video/BV1LQ4y127n4