7. RabbitMQ 消息队列——延时队列(Spring Boot + 安装message_exchange"耽 ...

7. RabbitMQ 消息队列——延时队列(Spring Boot + 安装message_exchange"耽误插件" 的详细配置说明)的详细讲解

@
目次

• 7. RabbitMQ 消息队列——延时队列(Spring Boot + 安装message_exchange"耽误插件" 的详细配置说明)的详细讲解
  
• 1. RabbitMQ 延时队列概述
  
• 2. RabbitMQ 中的 TTL
  
• 3.  RabbitMQ 整合 Spring Boot
  
  
  
  • 3.1  Spring Boot 当中搭建配置 RabbitMQ 的 队列 TTL(延时队列)
    
  
  
• null
  
  
  
  • 3.2 延时队列优化
    
  • 3.3 Rabbitmq 插件实现耽误队列
    
  
  
• 4. 补充：Docker 容器当中安装 message_exchange"耽误插件" 的详细步调
  
• 5. 末了：
  

1. RabbitMQ 延时队列概述

延时队列,队列内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素是希望 在指定时间到了以后或之前取出和处理，简朴来说，延时队列就是用来存放需要在指定时间被处理的 元素的队列。
延时队列的应用场景：

场景：有一个订单，15 分钟内如果不支付，就把该订单设置为交易关闭，那么就不能再举行支付了，这类实现耽误使命的场景就可以采用延时队列的方式来实现，固然除了延时队列，还可以采用其它的方式——> 定时使命的方式。

• 订单在非常钟之内未支付则自动取消。
  
• 新创建的店肆，如果在十天内都没有上传过商品，则自动发送消息提醒。
  
• 用户注册后，如果三天内没有登陆则举行短信提醒。
  
• 用户发起退款，如果三天内没有得到处理则通知相关运营职员。
  
• 预定会议后，需要在预定的时间点前 10 分钟，通知各个参会职员参加会议。
  

这些场景都有一个特点，需要在某个事件发生之后大概之前的指定时间点完成某一项使命，如： 发生订单天生事件，在非常钟之后查抄该订单支付状态，然后将未支付的订单举行关闭；看起来好像 使用定时使命，一直轮询数据，每秒查一次，取出需要被处理的数据，然后处理不就完事了吗？如果 数据量比较少，确实可以这样做，比如：对于“如果账单一周内未支付则举行自动结算”这样的需求， 如果对于时间不是严格限定，而是宽松意义上的一周，那么每天晚上跑个定时使命查抄一下所有未支 付的账单，确实也是一个可行的方案。但对于数据量比较大，而且时效性较强的场景，如：“订单十 分钟内未支付则关闭“，短期内未支付的订单数据可能会有许多，运动期间甚至会到达百万甚至万万 级别，对这么庞大的数据量仍旧使用轮询的方式显然是不可取的，很可能在一秒内无法完成所有订单 的查抄，同时会给数据库带来很大压力，无法满意业务要求而且性能低下。

2. RabbitMQ 中的 TTL

TTL 是什么呢？TTL 是 RabbitMQ 中一个消息大概队列的属性，表明一条消息大概队列中的所有消息的最大存活时间。
单位是毫秒。换句话说，如果一条消息设置了 TTL 属性大概进入了设置 TTL 属性的队列，那么这条消息如果在 TTL 设置的时间内没有被消费，则会成为 “死信 ” 。如果同时配置了队列的 TTL 和消息的 TTL ，那么较小的谁人值将会被使用，有两种方式设置了 TTL 。
消息设置 TTL：
另一种方式便是针对每条消息设置 TTL

队列设置 TTL：
第一种是在创建队列的时候设置队列的 “x-message-ttl ” 属性。

延时队列 和 死信队列：
如果设置了队列的 TTL 属性，那么一旦消息逾期，就会被队列抛弃(如果配置了死信队列被丢到死信队 列中)，而第二种方式，消息即使逾期，也不愿定会被马上抛弃，因为消息是否逾期是在即将投递到消费者 之前判定的，如果当前队列有严重的消息积压环境，则已逾期的消息大概还能存活较长时间；别的，还需 要注意的一点是，如果不设置 TTL，表示消息永久不会逾期，如果将 TTL 设置为 0，则表示除非此时可以 直接投递该消息到消费者，否则该消息将会被抛弃。
上一篇我们介绍了死信队列，刚刚又介绍了 TTL，至此利用 RabbitMQ 实现延时队列的两大要素已 经集齐，接下来只需要将它们举行融合，再加入一点点调味料，延时队列就可以新鲜出炉了。想想看，延 时队列，不就是想要消息耽误多久被处理吗，TTL 则刚好能让消息在耽误多久之后成为死信，另一方面， 成为死信的消息都会被投递到死信队列里，这样只需要消费者一直消费死信队列里的消息就完事了，因为 内里的消息都是希望被立即处理的消息。
3.  RabbitMQ 整合 Spring Boot

1. 创建项目

添加相关依赖，在 pom.xml 文件当中

1.     <dependencies>
2.         <dependency>
3.             <groupId>org.springframework.boot</groupId>
4.             <artifactId>spring-boot-starter</artifactId>
5.         </dependency>
7.         <dependency>
8.             <groupId>org.springframework.boot</groupId>
9.             <artifactId>spring-boot-starter-test</artifactId>
10.             <scope>test</scope>
11.         </dependency>
13.         
14.         <dependency>
15.             <groupId>org.springframework.boot</groupId>
16.             <artifactId>spring-boot-starter-amqp</artifactId>
17.         </dependency>
18.         <dependency>
19.             <groupId>org.springframework.boot</groupId>
20.             <artifactId>spring-boot-starter-web</artifactId>
21.         </dependency>
23.         <dependency>
24.             <groupId>com.alibaba</groupId>
25.             <artifactId>fastjson</artifactId>
26.             <version>1.2.47</version>
27.         </dependency>
28.         <dependency>
29.             <groupId>org.projectlombok</groupId>
30.             <artifactId>lombok</artifactId>
31.         </dependency>
32.         
33.         <dependency>
34.             <groupId>io.springfox</groupId>
35.             <artifactId>springfox-swagger2</artifactId>
36.             <version>2.9.2</version>
37.         </dependency>
38.         <dependency>
39.             <groupId>io.springfox</groupId>
40.             <artifactId>springfox-swagger-ui</artifactId>
41.             <version>2.9.2</version>
42.         </dependency>
43.         
44.         <dependency>
45.             <groupId>org.springframework.amqp</groupId>
46.             <artifactId>spring-rabbit-test</artifactId>
47.             <scope>test</scope>
48.         </dependency>
49.     </dependencies>

复制代码

1.     4.0.0            org.springframework.boot        spring-boot-starter-parent        2.5.3                 com.rainbowsea    springboot-rabbitmq    0.0.1-SNAPSHOT    springboot-rabbitmq    Demo project for Spring Boot    <dependencies>
2.         <dependency>
3.             <groupId>org.springframework.boot</groupId>
4.             <artifactId>spring-boot-starter</artifactId>
5.         </dependency>
7.         <dependency>
8.             <groupId>org.springframework.boot</groupId>
9.             <artifactId>spring-boot-starter-test</artifactId>
10.             <scope>test</scope>
11.         </dependency>
13.         
14.         <dependency>
15.             <groupId>org.springframework.boot</groupId>
16.             <artifactId>spring-boot-starter-amqp</artifactId>
17.         </dependency>
18.         <dependency>
19.             <groupId>org.springframework.boot</groupId>
20.             <artifactId>spring-boot-starter-web</artifactId>
21.         </dependency>
23.         <dependency>
24.             <groupId>com.alibaba</groupId>
25.             <artifactId>fastjson</artifactId>
26.             <version>1.2.47</version>
27.         </dependency>
28.         <dependency>
29.             <groupId>org.projectlombok</groupId>
30.             <artifactId>lombok</artifactId>
31.         </dependency>
32.         
33.         <dependency>
34.             <groupId>io.springfox</groupId>
35.             <artifactId>springfox-swagger2</artifactId>
36.             <version>2.9.2</version>
37.         </dependency>
38.         <dependency>
39.             <groupId>io.springfox</groupId>
40.             <artifactId>springfox-swagger-ui</artifactId>
41.             <version>2.9.2</version>
42.         </dependency>
43.         
44.         <dependency>
45.             <groupId>org.springframework.amqp</groupId>
46.             <artifactId>spring-rabbit-test</artifactId>
47.             <scope>test</scope>
48.         </dependency>
49.     </dependencies>

复制代码

在 resources/创建一个 application.properties 文件当中(固然也可以使用 yaml 配置文件)，配置相关的配置文件

1. spring.rabbitmq.host=192.168.76.156
2. spring.rabbitmq.port=5672
3. spring.rabbitmq.username=admin
4. spring.rabbitmq.password=123

复制代码

添加 Swagger 配置类

1. package com.rainbowsea.rabbitmq.springbootrabbitmq.config;
3. import org.springframework.context.annotation.Bean;
4. import org.springframework.context.annotation.Configuration;
5. import springfox.documentation.builders.ApiInfoBuilder;
6. import springfox.documentation.service.ApiInfo;
7. import springfox.documentation.service.Contact;
8. import springfox.documentation.spi.DocumentationType;
9. import springfox.documentation.spring.web.plugins.Docket;
10. import springfox.documentation.swagger2.annotations.EnableSwagger2;
12. @Configuration
13. @EnableSwagger2
14. public class SwaggerConfig {
15.     @Bean
16.     public Docket webApiConfig() {
17.         return new Docket(DocumentationType.SWAGGER_2)
18.                 .groupName("webApi")
19.                 .apiInfo(webApiInfo())
20.                 .select()
21.                 .build();
22.     }
24.     private ApiInfo webApiInfo() {
25.         return new ApiInfoBuilder()
26.                 .title("rabbitmq 接口文档")
27.                 .description("本文档描述了 rabbitmq 微服务接口定义")
28.                 .version("1.0")
29.                 .contact(new Contact("enjoy6288", "http://rainbowsea.com", "3339966@qq.com"))
30.                 .build();
31.     }
32. }

复制代码

3.1  Spring Boot 当中搭建配置 RabbitMQ 的 队列 TTL(延时队列)

代码架构图：
创建两个队列 QA 和 QB，两者队列 TTL 分别设置为 10S 和 40S，然后在创建一个交换机 X 和死信交 换机 Y，它们的类型都是 direct，创建一个死信队列 QD，它们的绑定关系如下：

1. 创建文件类代码（该代码就是用于配置 RabbitMQ 当中的交换机和队列的信息配置的），作为一个配置类存在的 。

1. import org.springframework.amqp.core.Binding;
2. import org.springframework.amqp.core.BindingBuilder;
3. import org.springframework.amqp.core.DirectExchange;
4. import org.springframework.amqp.core.Queue;
5. import org.springframework.amqp.core.QueueBuilder;
6. import org.springframework.beans.factory.annotation.Qualifier;
7. import org.springframework.context.annotation.Bean;
8. import org.springframework.context.annotation.Configuration;
10. import java.util.HashMap;
11. import java.util.Map;
13. @Configuration  // 标注是一个配置类，加载到 IOC 容器当中
14. public class TtlQueueConfig {
16.     // 普通交换机的名称
17.     public static final String X_EXCHANGE = "X";
19.     // 声明 XExchange 别名；会注入到 IOC 容器当中 普通交换机的
20.     @Bean(value = "xExchange")
21.     public DirectExchange xExchange() {
22.         return new DirectExchange(X_EXCHANGE);
23.     }
25. }

复制代码

1. package com.rainbowsea.rabbitmq.springbootrabbitmq.config;
4. import org.springframework.amqp.core.Binding;
5. import org.springframework.amqp.core.BindingBuilder;
6. import org.springframework.amqp.core.DirectExchange;
7. import org.springframework.amqp.core.Queue;
8. import org.springframework.amqp.core.QueueBuilder;
9. import org.springframework.beans.factory.annotation.Qualifier;
10. import org.springframework.context.annotation.Bean;
11. import org.springframework.context.annotation.Configuration;
13. import java.util.HashMap;
14. import java.util.Map;
16. @Configuration  // 标注是一个配置类，加载到 IOC 容器当中
17. public class TtlQueueConfig {
19.     // 普通交换机的名称
20.     public static final String X_EXCHANGE = "X";
21.     // 死信队列的名称
22.     public static final String DEAD_LETTER_QUEUE = "QD";
24.     // 死信交换机的名称
25.     public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
26.     // 普通队列的名称
27.     public static final String QUEUE_A = "QA";
28.     public static final String QUEUE_B = "QB";
30.     // 声明普通队列 TTL 为 10s
31.     @Bean("queueA")
32.     public Queue queueA() {
33.         // 创建一个 Map 用于存放，设置队列信息
34.         Map<String, Object> arguments = new HashMap<>(3);
35.         // 设置死信交换机 注意这里的参数 x-dead-letter-exchange 是固定的不可
36.         arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
37.         // 设置死信的 RoutingKey 为 YD ，注意这里的参数 x-dead-letter-routing-key 是固定的不可随便写
38.         arguments.put("x-dead-letter-routing-key", "YD");
39.         // 设置 TTL ，单位是 ms 注意这里的参数 x-message-ttl 是固定的不可随便写
40.         arguments.put("x-message-ttl", 10000);
42.         return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
43.     }
46.     // 声明普通队列 TTL 为 40s
47.     @Bean("queueB")
48.     public Queue queueB() {
49.         // 创建一个 Map 用于存放，设置队列信息
50.         Map<String, Object> arguments = new HashMap<>(3);
51.         // 设置死信交换机 注意这里的参数 x-dead-letter-exchange 是固定的不可
52.         arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
53.         // 设置死信的 RoutingKey 为 YD ，注意这里的参数 x-dead-letter-routing-key 是固定的不可随便写
54.         arguments.put("x-dead-letter-routing-key", "YD");
55.         // 设置 TTL ，单位是 ms 注意这里的参数 x-message-ttl 是固定的不可随便写
56.         arguments.put("x-message-ttl", 40000);
58.         return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
59.     }
62. }

复制代码

1. // 声明死信队列
2.     @Bean("queueD")
3.     public Queue queueD() {
4.         //  不带参数，就不需要 .withArguments(arguments)
5.         //return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
6.         return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
7.     }

复制代码

1. // 绑定
3.     /**
4.      * 注意：这里的  public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
5.      *                                   @Qualifier("xExchange") DirectExchange xExchange)
6.      *
7.      * @Qualifier("xExchange") DirectExchange xExchange)
8.      * @Qualifier("xExchange") DirectExchange xExchange)
9.      * @Qualifier 参数内的值是，我们上边：  @Bean(value = "xExchange")，@Bean("queueA") 注入到 IOC 容器当中的值
10.      * ，必须要是注入到了 IOC容器当中，不然是无法被使用上的，同时注意名称，不要写错了，要保持一致
11.      *  @Bean(value = "xExchange")
12.      *     public DirectExchange xExchange() {
13.      * @Bean("queueA") public Queue queueA() {
14.      */
15.     @Bean
16.     public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
17.                                   @Qualifier("xExchange") DirectExchange xExchange) {
18.         // 第一个参数是 队列名(注入到了IOC容器当中了)，第二个参数是交换机(注入到了IOC容器当中了)
19.         // 第三个蚕食是: 队列名对于绑定交换机之间的交互的  routingKey 信息"XA"
20.         // 表示将 队列 queueA 与 xExchange 交换机，通过 routingKey 为 "XA" 进行一个绑定
21.         return BindingBuilder.bind(queueA).to(xExchange).with("XA");
22.     }

复制代码

完整配置代码如下：

1. package com.rainbowsea.rabbitmq.springbootrabbitmq.config;import org.springframework.amqp.core.Binding;import org.springframework.amqp.core.BindingBuilder;import org.springframework.amqp.core.DirectExchange;import org.springframework.amqp.core.Queue;import org.springframework.amqp.core.QueueBuilder;import org.springframework.beans.factory.annotation.Qualifier;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import java.util.HashMap;import java.util.Map;@Configuration  // 标注是一个配置类，加载到 IOC 容器当中public class TtlQueueConfig {    // 普通交换机的名称    public static final String X_EXCHANGE = "X";    // 死信队列的名称    public static final String DEAD_LETTER_QUEUE = "QD";    // 死信交换机的名称    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";    // 普通队列的名称    public static final String QUEUE_A = "QA";    public static final String QUEUE_B = "QB";    // 声明 XExchange 别名；会注入到 IOC 容器当中 普通交换机的    @Bean(value = "xExchange")    public DirectExchange xExchange() {        return new DirectExchange(X_EXCHANGE);    }    // 声明 yExchange 别名；会注入到 IOC 容器当中 普通交换机的    @Bean(value = "yExchange")    public DirectExchange yExchange() {        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);    }    // 声明普通队列 TTL 为 10s    @Bean("queueA")    public Queue queueA() {        // 创建一个 Map 用于存放，设置队列信息        Map arguments = new HashMap(3);        // 设置死信交换机 注意这里的参数 x-dead-letter-exchange 是固定的不可        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);        // 设置死信的 RoutingKey 为 YD ，注意这里的参数 x-dead-letter-routing-key 是固定的不可随便写        arguments.put("x-dead-letter-routing-key", "YD");        // 设置 TTL ，单位是 ms 注意这里的参数 x-message-ttl 是固定的不可随便写        arguments.put("x-message-ttl", 10000);        return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();    }    // 声明普通队列 TTL 为 40s    @Bean("queueB")    public Queue queueB() {        // 创建一个 Map 用于存放，设置队列信息        Map arguments = new HashMap(3);        // 设置死信交换机 注意这里的参数 x-dead-letter-exchange 是固定的不可        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);        // 设置死信的 RoutingKey 为 YD ，注意这里的参数 x-dead-letter-routing-key 是固定的不可随便写        arguments.put("x-dead-letter-routing-key", "YD");        // 设置 TTL ，单位是 ms 注意这里的参数 x-message-ttl 是固定的不可随便写        arguments.put("x-message-ttl", 40000);        return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();    }    // 声明死信队列
2.     @Bean("queueD")
3.     public Queue queueD() {
4.         //  不带参数，就不需要 .withArguments(arguments)
5.         //return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
6.         return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
7.     }    // 绑定    /**     * 注意：这里的  public Binding queueABindingX(@Qualifier("queueA") Queue queueA,     *                                   @Qualifier("xExchange") DirectExchange xExchange)     *     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier 参数内的值是，我们上边：  @Bean(value = "xExchange")，@Bean("queueA") 注入到 IOC 容器当中的值     * ，必须要是注入到了 IOC容器当中，不然是无法被使用上的，同时注意名称，不要写错了，要保持一致     *  @Bean(value = "xExchange")     *     public DirectExchange xExchange() {     * @Bean("queueA") public Queue queueA() {     */    @Bean  // 绑定也是要注入到 IOC 容器当中的    public Binding queueABindingX(@Qualifier("queueA") Queue queueA,                                  @Qualifier("xExchange") DirectExchange xExchange) {        // 第一个参数是 队列名(注入到了IOC容器当中了)，第二个参数是交换机(注入到了IOC容器当中了)        // 第三个蚕食是: 队列名对于绑定交换机之间的交互的  routingKey 信息"XA"        // 表示将 队列 queueA 与 xExchange 交换机，通过 routingKey 为 "XA" 举行一个绑定        return BindingBuilder.bind(queueA).to(xExchange).with("XA");    }    // 绑定    /**     * 注意：这里的  public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,     *                                   @Qualifier("xExchange") DirectExchange xExchange)     *     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier 参数内的值是，我们上边：  @Bean(value = "xExchange")，@Bean("queueB") 注入到 IOC 容器当中的值     * ，必须要是注入到了 IOC容器当中，不然是无法被使用上的，同时注意名称，不要写错了，要保持一致     *  @Bean(value = "xExchange")     *     public DirectExchange xExchange() {     * @Bean("queueB") public Queue queueB() {     */    @Bean  // 绑定也是要注入到 IOC 容器当中的    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,                                  @Qualifier("xExchange") DirectExchange xExchange) {        // 第一个参数是 队列名(注入到了IOC容器当中了)，第二个参数是交换机(注入到了IOC容器当中了)        // 第三个蚕食是: 队列名对于绑定交换机之间的交互的  routingKey 信息"XB"        // 表示将 队列 queueB 与 xExchange 交换机，通过 routingKey 为 "XB" 举行一个绑定        return BindingBuilder.bind(queueB).to(xExchange).with("XB");    }    // 绑定    /**     * 注意：这里的  public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,     *                                   @Qualifier("yExchange") DirectExchange yExchange)     *     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier 参数内的值是，我们上边：  @Bean(value = "yExchange")，@Bean("queueD") 注入到 IOC 容器当中的值     * ，必须要是注入到了 IOC容器当中，不然是无法被使用上的，同时注意名称，不要写错了，要保持一致     * @Bean(value = "yExchange")     * public DirectExchange yExchange() {     * @Bean("queueD")     *     public Queue queueD() {     */    @Bean // 绑定也是要注入到 IOC 容器当中的    public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,                                  @Qualifier("yExchange") DirectExchange yExchange) {        // 第一个参数是 队列名(注入到了IOC容器当中了)，第二个参数是交换机(注入到了IOC容器当中了)        // 第三个蚕食是: 队列名对于绑定交换机之间的交互的  routingKey 信息"YD"        // 表示将 队列 queueD 与 yExchange 交换机，通过 routingKey 为 "YD" 举行一个绑定        return BindingBuilder.bind(queueD).to(yExchange).with("YD");    }}

复制代码

生产者——》发送消息的代码：

1. package com.rainbowsea.rabbitmq.springbootrabbitmq.controller;
4. import lombok.extern.slf4j.Slf4j;
5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
6. import org.springframework.web.bind.annotation.GetMapping;
7. import org.springframework.web.bind.annotation.PathVariable;
8. import org.springframework.web.bind.annotation.RequestMapping;
9. import org.springframework.web.bind.annotation.RestController;
11. import javax.annotation.Resource;
12. import java.util.Date;
14. /**
15. * 生产者：发送延时消息
16. * http://localhost:8080/ttl/sendMsg/嘻嘻嘻
17. */
20. @RestController
21. @RequestMapping("/ttl")
22. @Slf4j
23. public class SendMsgController {
25.     @Resource  // 注入到 IOC 容器当中
26.     private RabbitTemplate rabbitTemplate;
29.     @GetMapping("/sendMsg/{message}")
30.     public void sendMsg(@PathVariable String message) {
31.         log.info("当前时间:{},发送一条消息给两个 TTL 队列:{}", new Date().toString(), message);
32.         rabbitTemplate.convertAndSend("X","XA","消息来自ttl为 10s的队列" + message);
33.         rabbitTemplate.convertAndSend("X","XB","消息来自ttl为 40s的队列" + message);
35.     }
37. }

复制代码

消费者——》消费/读取消息的代码：

1. package com.rainbowsea.rabbitmq.springbootrabbitmq.consumer;
4. import com.rabbitmq.client.Channel;
5. import lombok.extern.slf4j.Slf4j;
6. import org.springframework.amqp.core.Message;
8. import org.springframework.amqp.rabbit.annotation.RabbitListener;
9. import org.springframework.stereotype.Component;
11. import java.io.IOException;
12. import java.util.Date;
14. /**
15. * 消费者(消费/读取消息)
16. */
18. @Slf4j
19. @Component  // 当 IOC 容器读取到
20. public class DeadLetterQueueConsumer {
23.     @RabbitListener(queues = "QD")
24.     public void receiveD(Message message, Channel channel) {
25.         String msg = new String(message.getBody());
26.         log.info("当前时间:{},收到死信队列的消息:{}", new Date().toString(), msg);
27.     }
30. }

复制代码

运行效果：

该延时队列存在的标题：
第一条消息在 10 S 后变成了死信消息，然后被消费者消费/读取掉了，第二条消息在 40S 之后变成了死信消息，然后被消费掉，这样一个延时队列就被打造完成了。
不过，该延时队列存在一个标题：

如果这样使用的话，岂不是每增长一个新的时间需求，就要新增一个队列。 这里只有 10S  和 40S 两个时间选项，如果需要一个小时后处理，那么就需要增长 TTL 为一个小时的队列，如果是预定会议室，然后提前通知这样的场景，岂不是要增长无数个队列才能满意需求。

3.2 延时队列优化

优化思绪： 就是把延时时间，交给发送消息的生产者上决定，而不是写死在队列当中，通过一个无延时队列来处理，生产者上设置的一个延时发送的时间。

在这里新增了一个队列 QC,绑定关系如下，该队列不设置 TTL 时间

特别注意： 这里需要将上面创建了的 交换机，队列删除了，因为 RabbitMQ 消息队列不能将一个已经存在的交换机/队列，的基础上，举行修改，而是只能重新天生 。

在 TtlQueueConfig 配置类上添加上一个，RoutingKey 为 XC ，队列为 QC，对应 死信队列还是 QD 。

1. import org.springframework.amqp.core.Binding;
2. import org.springframework.amqp.core.BindingBuilder;
3. import org.springframework.amqp.core.DirectExchange;
4. import org.springframework.amqp.core.Queue;
5. import org.springframework.amqp.core.QueueBuilder;
6. import org.springframework.beans.factory.annotation.Qualifier;
7. import org.springframework.context.annotation.Bean;
8. import org.springframework.context.annotation.Configuration;
10. import java.util.HashMap;
11. import java.util.Map;
13. @Configuration  // 标注是一个配置类，加载到 IOC 容器当中
14. public class TtlQueueConfig {
16.     // 普通交换机的名称
17.     public static final String X_EXCHANGE = "X";
18.     // 死信队列的名称
19.     public static final String DEAD_LETTER_QUEUE = "QD";
21.     // 死信交换机的名称
22.     public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
23.     // 普通队列的名称
24.     public static final String QUEUE_A = "QA";
25.     public static final String QUEUE_B = "QB";
29.     //普通队列的名称
30.     public static final String QUEUE_C = "QC";
31.     // 声明队列 QC 同时绑定上 (以 RoutingKey 为 XC  )QD 死信交换机
32.     @Bean("queueC")
33.     public Queue queueC(){
34.         // 创建一个 Map 用于存放，设置队列信息
35.         Map<String, Object> arguments = new HashMap<>(3);
36.         // 设置死信交换机 注意这里的参数 x-dead-letter-exchange 是固定的不可
37.         arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
38.         // 设置死信的 RoutingKey 为 YD ，注意这里的参数 x-dead-letter-routing-key 是固定的不可随便写
39.         arguments.put("x-dead-letter-routing-key", "YD");
40.         // 不设置 TTL 时间，也是通过生产者来设置
41.         return QueueBuilder.durable(QUEUE_C).withArguments(arguments).build();
42.     }
44.     // 将声明的队列 QC  绑定到 X 交换机当中
45.     @Bean  // 绑定也是要注入到 IOC 容器当中的
46.     public Binding queueBindingX(@Qualifier("queueC") Queue queueC,
47.                                   @Qualifier("xExchange") DirectExchange xExchange) {
48.         // 第一个参数是 队列名(注入到了IOC容器当中了)，第二个参数是交换机(注入到了IOC容器当中了)
49.         // 第三个蚕食是: 队列名对于绑定交换机之间的交互的  routingKey 信息"XC"
50.         // 表示将 队列 queueC 与 xExchange 交换机，通过 routingKey 为 "XC" 进行一个绑定
51.         return BindingBuilder.bind(queueC).to(xExchange).with("XC");
52.     }
54. }

复制代码

完整代码内容：

1. package com.rainbowsea.rabbitmq.springbootrabbitmq.config;import org.springframework.amqp.core.Binding;import org.springframework.amqp.core.BindingBuilder;import org.springframework.amqp.core.DirectExchange;import org.springframework.amqp.core.Queue;import org.springframework.amqp.core.QueueBuilder;import org.springframework.beans.factory.annotation.Qualifier;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import java.util.HashMap;import java.util.Map;@Configuration  // 标注是一个配置类，加载到 IOC 容器当中public class TtlQueueConfig {    // 普通交换机的名称    public static final String X_EXCHANGE = "X";    // 死信队列的名称    public static final String DEAD_LETTER_QUEUE = "QD";    // 死信交换机的名称    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";    // 普通队列的名称    public static final String QUEUE_A = "QA";    public static final String QUEUE_B = "QB";    //普通队列的名称    public static final String QUEUE_C = "QC";    // 声明队列 QC 同时绑定上 (以 RoutingKey 为 XC  )QD 死信交换机    @Bean("queueC")    public Queue queueC(){        // 创建一个 Map 用于存放，设置队列信息        Map arguments = new HashMap(3);        // 设置死信交换机 注意这里的参数 x-dead-letter-exchange 是固定的不可        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);        // 设置死信的 RoutingKey 为 YD ，注意这里的参数 x-dead-letter-routing-key 是固定的不可随便写        arguments.put("x-dead-letter-routing-key", "YD");        // 不设置 TTL 时间，也是通过生产者来设置        return QueueBuilder.durable(QUEUE_C).withArguments(arguments).build();    }    // 将声明的队列 QC  绑定到 X 交换机当中    @Bean  // 绑定也是要注入到 IOC 容器当中的    public Binding queueBindingX(@Qualifier("queueC") Queue queueC,                                  @Qualifier("xExchange") DirectExchange xExchange) {        // 第一个参数是 队列名(注入到了IOC容器当中了)，第二个参数是交换机(注入到了IOC容器当中了)        // 第三个蚕食是: 队列名对于绑定交换机之间的交互的  routingKey 信息"XC"        // 表示将 队列 queueC 与 xExchange 交换机，通过 routingKey 为 "XC" 举行一个绑定        return BindingBuilder.bind(queueC).to(xExchange).with("XC");    }    // 声明 XExchange 别名；会注入到 IOC 容器当中 普通交换机的    @Bean(value = "xExchange")    public DirectExchange xExchange() {        return new DirectExchange(X_EXCHANGE);    }    // 声明 yExchange 别名；会注入到 IOC 容器当中 普通交换机的    @Bean(value = "yExchange")    public DirectExchange yExchange() {        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);    }    // 声明普通队列 TTL 为 10s    @Bean("queueA")    public Queue queueA() {        // 创建一个 Map 用于存放，设置队列信息        Map arguments = new HashMap(3);        // 设置死信交换机 注意这里的参数 x-dead-letter-exchange 是固定的不可        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);        // 设置死信的 RoutingKey 为 YD ，注意这里的参数 x-dead-letter-routing-key 是固定的不可随便写        arguments.put("x-dead-letter-routing-key", "YD");        // 设置 TTL ，单位是 ms 注意这里的参数 x-message-ttl 是固定的不可随便写        arguments.put("x-message-ttl", 10000);        return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();    }    // 声明普通队列 TTL 为 40s    @Bean("queueB")    public Queue queueB() {        // 创建一个 Map 用于存放，设置队列信息        Map arguments = new HashMap(3);        // 设置死信交换机 注意这里的参数 x-dead-letter-exchange 是固定的不可        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);        // 设置死信的 RoutingKey 为 YD ，注意这里的参数 x-dead-letter-routing-key 是固定的不可随便写        arguments.put("x-dead-letter-routing-key", "YD");        // 设置 TTL ，单位是 ms 注意这里的参数 x-message-ttl 是固定的不可随便写        arguments.put("x-message-ttl", 40000);        return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();    }    // 声明死信队列
2.     @Bean("queueD")
3.     public Queue queueD() {
4.         //  不带参数，就不需要 .withArguments(arguments)
5.         //return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
6.         return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
7.     }    // 绑定    /**     * 注意：这里的  public Binding queueABindingX(@Qualifier("queueA") Queue queueA,     *                                   @Qualifier("xExchange") DirectExchange xExchange)     *     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier 参数内的值是，我们上边：  @Bean(value = "xExchange")，@Bean("queueA") 注入到 IOC 容器当中的值     * ，必须要是注入到了 IOC容器当中，不然是无法被使用上的，同时注意名称，不要写错了，要保持一致     *  @Bean(value = "xExchange")     *     public DirectExchange xExchange() {     * @Bean("queueA") public Queue queueA() {     */    @Bean  // 绑定也是要注入到 IOC 容器当中的    public Binding queueABindingX(@Qualifier("queueA") Queue queueA,                                  @Qualifier("xExchange") DirectExchange xExchange) {        // 第一个参数是 队列名(注入到了IOC容器当中了)，第二个参数是交换机(注入到了IOC容器当中了)        // 第三个蚕食是: 队列名对于绑定交换机之间的交互的  routingKey 信息"XA"        // 表示将 队列 queueA 与 xExchange 交换机，通过 routingKey 为 "XA" 举行一个绑定        return BindingBuilder.bind(queueA).to(xExchange).with("XA");    }    // 绑定    /**     * 注意：这里的  public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,     *                                   @Qualifier("xExchange") DirectExchange xExchange)     *     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier 参数内的值是，我们上边：  @Bean(value = "xExchange")，@Bean("queueB") 注入到 IOC 容器当中的值     * ，必须要是注入到了 IOC容器当中，不然是无法被使用上的，同时注意名称，不要写错了，要保持一致     *  @Bean(value = "xExchange")     *     public DirectExchange xExchange() {     * @Bean("queueB") public Queue queueB() {     */    @Bean  // 绑定也是要注入到 IOC 容器当中的    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,                                  @Qualifier("xExchange") DirectExchange xExchange) {        // 第一个参数是 队列名(注入到了IOC容器当中了)，第二个参数是交换机(注入到了IOC容器当中了)        // 第三个蚕食是: 队列名对于绑定交换机之间的交互的  routingKey 信息"XB"        // 表示将 队列 queueB 与 xExchange 交换机，通过 routingKey 为 "XB" 举行一个绑定        return BindingBuilder.bind(queueB).to(xExchange).with("XB");    }    // 绑定    /**     * 注意：这里的  public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,     *                                   @Qualifier("yExchange") DirectExchange yExchange)     *     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier("xExchange") DirectExchange xExchange)     * @Qualifier 参数内的值是，我们上边：  @Bean(value = "yExchange")，@Bean("queueD") 注入到 IOC 容器当中的值     * ，必须要是注入到了 IOC容器当中，不然是无法被使用上的，同时注意名称，不要写错了，要保持一致     * @Bean(value = "yExchange")     * public DirectExchange yExchange() {     * @Bean("queueD")     *     public Queue queueD() {     */    @Bean // 绑定也是要注入到 IOC 容器当中的    public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,                                  @Qualifier("yExchange") DirectExchange yExchange) {        // 第一个参数是 队列名(注入到了IOC容器当中了)，第二个参数是交换机(注入到了IOC容器当中了)        // 第三个蚕食是: 队列名对于绑定交换机之间的交互的  routingKey 信息"YD"        // 表示将 队列 queueD 与 yExchange 交换机，通过 routingKey 为 "YD" 举行一个绑定        return BindingBuilder.bind(queueD).to(yExchange).with("YD");    }}

复制代码

生产者——> 发送消息，代码

这里，生产者，指定一个延时时间(TTL)，交给 X 交换机——>以 routingKey = XC 发送给 QC 队列——> 发送给死信交换机——> 以 routingKey = YD 分配给 QD 死信队列——>末了给消费者读取/消费消息。

1. import lombok.extern.slf4j.Slf4j;
2. import org.springframework.amqp.rabbit.core.RabbitTemplate;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.web.bind.annotation.GetMapping;
5. import org.springframework.web.bind.annotation.PathVariable;
6. import org.springframework.web.bind.annotation.RequestMapping;
7. import org.springframework.web.bind.annotation.RestController;
9. import javax.annotation.Resource;
10. import java.util.Date;
12. /**
13. * 生产者：发送延时消息
14. * http://localhost:8080/ttl/sendMsg/嘻嘻嘻
15. */
18. @RestController
19. @RequestMapping("ttl")
20. @Slf4j
21. public class SendMsgController {
23.     @Resource  // 注入到 IOC 容器当中
24.     private RabbitTemplate rabbitTemplate;
27.     @GetMapping("/sendExpirationMsg/{message}/{ttlTime}")
28.     public void sendMsg(@PathVariable String message,
29.                         @PathVariable String ttlTime) {
30.         log.info("当前时间:{},发送一条时长{}毫秒 TTL 消息给队列QC:{}", new Date().toString(), ttlTime, message);
31.         rabbitTemplate.convertAndSend("X", "XC", message, msg -> {
32.             // 发送消息的时候，延时时长
33.             msg.getMessageProperties().setExpiration(ttlTime);
34.             return msg;
35.         });
36.         log.info("当前时间：{},发送一条时长{}毫秒 TTL 信息给队列 C:{}", new Date(), ttlTime, message);
38.     }
39. }

复制代码

测试效果：
浏览器发送哀求：

• http://localhost:8080/ttl/sendExpirationMsg/你好1/20000
  
• http://localhost:8080/ttl/sendExpirationMsg/你好2/2000
  

分析测试效果：

看起来好像没什么标题，但是在最开始的时候，就介绍过如果使用消息属性上设置 TTL 的方式，消息可能并不会按时 “殒命” ，因为RabbitMQ 只会查抄第一个消息是否逾期，如果逾期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行 。

3.3 Rabbitmq 插件实现耽误队列

上文中提到的标题，确实是一个标题，如果不能实现在消息粒度上的 TTL，并使其在设置的 TTL 时间 及时殒命，就无法计划成一个通用的延时队列。那如何办理呢，接下来我们就去办理该标题。
我们可以通过安装一个：rabbitmq-delayed-message-exchange 插件 来办理这个标题。这个插件是如何办理的呢：既然延时队列，将消息放入到队列当中存在，一个消息逾期时间优先级的标题，那么我们就不将延时的消息，放入到队列当中，而是通过交换机举行处理。
rabbitmq-delayed-message-exchange 插件 就是让我们延时的消息，可以不走队列，而是通过交换机本身举行一个处理，让消费者直接从交换机当中消费，读取消息。

消息发送后不会直接投递到队列，
而是存储到 Mnesia（嵌入式数据库），查抄 x-delay 时间（消息头部）；
耽误插件在 RabbitMQ 3.5.7 及以上的版本才支持，依赖 Erlang/OPT 18.0 及以上运行环境；
Mnesia 是一个小型数据库，不得当于大量耽误消息的实现
办理了消息逾期时间不一致(优先级)出现的标题。
安装：安装延时队列插件 详细步调：

• 在官网上下载：https://www.rabbitmq.com/community-plugins
  

下载 rabbitmq_delayed_message_exchange 插件

• 然后解压放置到 RabbitMQ 的插件目次，进入 RabbitMQ 的安装目次下的 plgins 目次，执行下面命令让该插件见效。
  

1. [root@RainbowSea plugins]# cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins # 进入到 RabbitMQ 插件目录

复制代码

• 将下载好的 rabbitmq_delayed_message_exchange 插件，上传到 RabbitMQ 的安装目次下的 plgins 目次 当中。
  

• 执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 命令，启用该插件
  

1. [root@RainbowSea plugins]# rabbitmq-plugins enable rabbitmq_delayed_message_exchange

复制代码

• rabbitmq-plugins list 查看所有插件
  

1. [root@RainbowSea plugins]# rabbitmq-plugins list

复制代码

添加成功的效果：

详细代码实现：

• 在这里新增了一个队列 delayed.queue,一个自定义交换机 delayed.exchange，绑定关系如下:
  

编写配置类代码：
在我们自定义的交换机中，这是一种新的交换类型，该类型消息支持耽误投递机制 消息转达后并 不会立即投递到目标队列中，而是存储在 mnesia (一个分布式数据系统)表中，当到达投递时间时，才投递到目标队列中。

1. package com.rainbowsea.rabbitmq.springbootrabbitmq.config;
4. import org.springframework.amqp.core.Binding;
5. import org.springframework.amqp.core.BindingBuilder;
6. import org.springframework.amqp.core.CustomExchange;
7. import org.springframework.amqp.core.Queue;
8. import org.springframework.beans.factory.annotation.Qualifier;
9. import org.springframework.context.annotation.Bean;
10. import org.springframework.context.annotation.Configuration;
12. import java.util.HashMap;
13. import java.util.Map;
15. @Configuration
16. public class DelayedQueueConfig {
18.     public static final String DELAYED_QUEUE_NAME = "delayed.queue";
19.     public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
20.     public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
23.     @Bean("delayedQueue")  // 声明队列，基于 rabbitmq_delayed_message_exchange 插件的
24.     public Queue delayedQueue() {
25.         return new Queue(DELAYED_QUEUE_NAME);
26.     }
28.     // 声明交换机(可以算是一种自定义交换机)，基于 rabbitmq_delayed_message_exchange 插件的
29.     @Bean("delayedExchange")
30.     public CustomExchange delayedExchange() {
31.         // 创建一个 Map 用于存放，设置队列信息
32.         Map<String, Object> arguments = new HashMap<>(3);
33.         // 自定义交换机的类型, 这里定义为 direct 直接类型
34.         arguments.put("x-delayed-type", "direct");
35.         /**
36.          * CustomExchange 方法参数
37.          * 1.交换机的名称
38.          * 2.交换机的类型
39.          * 3.是否需要持久化
40.          * 4.是否需要自动删除
41.          * 5.其它的参数
42.          */
44.         return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, arguments);
45.     }
48.     // 基于 rabbitmq_delayed_message_exchange 插件的 将队列绑定到交换机当中
49.     @Bean
50.     public Binding delayedQueueBindingdelayedExchange(
51.             @Qualifier("delayedQueue") Queue delayedQueue,
52.             @Qualifier("delayedExchange") CustomExchange delayedExchange) {
53.         return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
54.     }
55. }

复制代码

生产者-发送消息的： 这里我们使用的是，通过生产者，设置延时时间，而不是通过队列写死延时时间的方式。

1. import com.rainbowsea.rabbitmq.springbootrabbitmq.config.DelayedQueueConfig;
2. import lombok.extern.slf4j.Slf4j;
3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
4. import org.springframework.beans.factory.annotation.Autowired;
5. import org.springframework.web.bind.annotation.GetMapping;
6. import org.springframework.web.bind.annotation.PathVariable;
7. import org.springframework.web.bind.annotation.RequestMapping;
8. import org.springframework.web.bind.annotation.RestController;
11. import javax.annotation.Resource;
12. import java.util.Date;
14. /**
15. * 生产者：发送延时消息
16. * http://localhost:8080/ttl/sendMsg/嘻嘻嘻
17. */
20. @RestController
21. @RequestMapping("ttl")
22. @Slf4j
23. public class SendMsgController {
25.     @Resource  // 注入到 IOC 容器当中
26.     private RabbitTemplate rabbitTemplate;
29.     @GetMapping("sendDelayMsg/{message}/{delayTime}")
30.     public void sendMsg(@PathVariable String message, @PathVariable Integer delayTime) {
31.         log.info(" 当前时间: {}, 发送一条延迟{} 毫秒的信息给队列 delayed.queue:{}", new Date(), delayTime, message);
32.         rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME, DelayedQueueConfig.DELAYED_ROUTING_KEY, message,
33.                 correlationData -> {
34.                     // 发送消息的时候，延迟时长，单位:ms
35.                     correlationData.getMessageProperties().setDelay(delayTime);
36.                     return correlationData;
37.                 });
38.     }
40. }

复制代码

消费者/读取/消费——使用插件交换机当中延时的消息：

1. package com.rainbowsea.rabbitmq.springbootrabbitmq.consumer;
4. import lombok.extern.slf4j.Slf4j;
5. import org.springframework.amqp.core.Message;
6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
7. import org.springframework.stereotype.Component;
8. import com.rainbowsea.rabbitmq.springbootrabbitmq.config.DelayedQueueConfig;
10. import java.util.Date;
13. /**
14. * 消费者，消费/读取消息，基于rabbitmq_delayed_message_exchange 插件
15. */
16. @Slf4j
17. @Component
18. public class DelayQueueConsumer {
20.     // 监听消息
21.     @RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
22.     public void receiveDelayQueue(Message message) {
23.         String msg = new String(message.getBody());
24.         log.info("当前时间:{} ,收到延时队列的消息:{}", new Date().toString(), msg);
25.     }
26. }

复制代码

运行效果：  浏览器当中发出哀求：

• http://localhost:8080/ttl/sendDelayMsg/come on baby1/20000
  
• http://localhost:8080/ttl/sendDelayMsg/come on baby2/2000
  

总结：
延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的利用 RabbitMQ 的特性，如：消息可靠发送，消息可靠投递，死信队列 来保障消息至少被消费一次以及未被正确处理的消息不会被抛弃。别的，通过 RabbitMQ 集群的特性，可以很好的办理单击故障标题，不会因为单个节点挂掉导致延时队列不可用大概消息丢失。
固然，延时队列还有许多其它选择，比如利用 Java 的 DelayQueue，利用 Redis的 zset，利用 Quartz 大概利用 kafka 的时间轮，这些方式各有特点，看需要适用的场景。
4. 补充：Docker 容器当中安装 message_exchange"耽误插件" 的详细步调

• 确定 Docker 容器对应映射的卷的目次位置：
  

1. docker inspect rabbitmq

复制代码

对应的运行效果：

1.         "Mounts": [
2.             {
3.                 "Type": "volume",
4.                 "Name": "rabbitmq-plugin",
5.                 "Source": "/var/lib/docker/volumes/rabbitmq-plugin/_data",
6.                 "Destination": "/plugins",
7.                 "Driver": "local",
8.                 "Mode": "z",
9.                 "RW": true,
10.                 "Propagation": ""
11.             },
12.             {
13.                 "Type": "volume",
14.                 "Name": "cca7bc3012f5b76bd6c47a49ca6911184f9076f5f6263b41f4b9434a7f269b11",
15.                 "Source": "/var/lib/docker/volumes/cca7bc3012f5b76bd6c47a49ca6911184f9076f5f6263b41f4b9434a7f269b11/_data",
16.                 "Destination": "/var/lib/rabbitmq",
17.                 "Driver": "local",
18.                 "Mode": "",
19.                 "RW": true,
20.                 "Propagation": ""
21.             }
22.         ]

复制代码

和容器内 /plugins 目次对应的宿主机目次是：/var/lib/docker/volumes/rabbitmq-plugin/_data

注意：这里必须必须必须，将下载到的 插件 拷贝到 RabbitMQ 服务器 plugins 目次下，不然时无法安装成功的。

• 下载对应的 messaging 延时插件
  

messaging 延时插件的官网文档地址： https://www.rabbitmq.com/community-plugins

下载在 Docker 容器当中下载安装文件：

1. wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v3.13.0/rabbitmq_delayed_message_exchange-3.13.0.ez
2. mv rabbitmq_delayed_message_exchange-3.13.0.ez /var/lib/docker/volumes/rabbitmq-plugin/_data

复制代码

• 启用该延时插件：
  

1. # 登录进入容器内部
2. docker exec -it rabbitmq /bin/bash
4. # rabbitmq-plugins命令所在目录已经配置到$PATH环境变量中了，可以直接调用
5. rabbitmq-plugins enable rabbitmq_delayed_message_exchange
7. # 退出Docker容器
8. exit
10. # 重启Docker容器
11. docker restart rabbitmq

复制代码

• 插件安装成功的，确定，验证：
  

• 确认点1：查看当前节点已启用插件的列表：
  

• 确认点2：如果创建新交换机时可以在type中看到x-delayed-message选项，那就说明插件安装好了
  

5. 末了：

“在这个末了的篇章中，我要表达我对每一位读者的感激之情。你们的关注和回复是我创作的动力源泉，我从你们身上吸取了无尽的灵感与勇气。我会将你们的鼓励留在心底，继续在其他的领域奋斗。感谢你们，我们总会在某个时刻再次相遇。”

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