定制化开发RabbitMQ延迟队列

目录

💌 介绍

顾名思义：定制化开发首先它要具有的特性，定制化开发再给它附加一个延迟消定制化开发费队列消息的功能，定制化开发也就是说可以指定队列定制化开发中的消息在哪个时间点被消费。

💒 使用场景

  • 定制化开发预支付订单创建成功后，30定制化开发分钟后还没有支付，定制化开发自动取消订单，定制化开发修改订单状态
  • 定制化开发用户注册成功后，如果3定制化开发天没有登录则进行短信提醒
  • 定制化开发优惠券过期前发送短信进行提醒
  • ....

定制化开发以上场景都可以用延时定制化开发队列来完成

🏳‍🌈 模拟案例

需求：定制化开发生产者发布消息，10秒、60秒后消费者拿到消息进行消费

📕 准备工作

导入依赖

  1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-amqp</artifactId>
  4. </dependency>

 配置RabbitMQ连接相关信息

  1. #MySQL
  2. spring:
  3.   rabbitmq:
  4.     host: 127.0.0.1
  5.     port: 5672 
  6.     username: xxxx
  7.     password: xxx
  8.  
  9. server:
  10.   port: 8087

🏴 写法一(死信队列TTL)

生产者生产消息——>到交换机分发给对应的队列(A10秒过期，B60秒过期)——>过期后到死信交换机——>消费者进行消费（执行顺序如下图）

 RabbitMQ配置文件

  1. import org.springframework.amqp.core.*;
  2. import org.springframework.beans.factory.annotation.Qualifier;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5.  
  6. import java.util.HashMap;
  7.  
  8. /**
  9.  * @author 小影
  10.  * @create: 2022/8/18 10:26
  11.  * @describe：mq配置 如示例图配置：2交换机、4队列、4路由key
  12.  */
  13. @Configuration
  14. public class RabbitMQConfiguration {
  15.    // 延迟交换机
  16.    public static final String DELAY_EXCHANGE_NAME = "delay.exchange";
  17.    // 延迟队列
  18.    public static final String DELAY_QUEUE_NAME_A = "delay.queue.a";
  19.    public static final String DELAY_QUEUE_NAME_B = "delay.queue.b";
  20.    // 延迟队列路由key
  21.    public static final String DELAY_QUEUE_ROUTING_KEY_A = "delay.routingKey.a";
  22.    public static final String DELAY_QUEUE_ROUTING_KEY_B = "delay.routingKey.b";
  23.  
  24.    // 死信交换机
  25.    public static final String DEAD_LETTER_EXCHANGE_NAME = "dead.letter.exchange";
  26.    // 死信队列
  27.    public static final String DEAD_LETTER_QUEUE_NAME_A = "dead.letter.queue.a";
  28.    public static final String DEAD_LETTER_QUEUE_NAME_B = "dead.letter.queue.b";
  29.    // 私信队列路由key
  30.    public static final String DEAD_LETTER_ROUTING_KEY_A = "dead.letter.delay_10s.routingkey.a";
  31.    public static final String DEAD_LETTER_ROUTING_KEY_B = "dead.letter.delay_60s.routingkey.b";
  32.  
  33.    // 声明延迟交换机
  34.    @Bean("delayExchange")
  35.    public DirectExchange delayExchange() {
  36.       return new DirectExchange(DELAY_EXCHANGE_NAME);
  37.    }
  38.  
  39.    // 声明死信交换机
  40.    @Bean("deadLetterExchange")
  41.    public DirectExchange deadLetterExchange() {
  42.       return new DirectExchange(DEAD_LETTER_EXCHANGE_NAME);
  43.    }
  44.  
  45.    // 声明延迟队列A，延迟10s，并且绑定到对应的死信交换机
  46.    @Bean("delayQueueA")
  47.    public Queue delayQueueA() {
  48.       HashMap<String, Object> args = new HashMap<>();
  49.       // 声明队列绑定的死信交换机
  50.       args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE_NAME);
  51.       // 声明队列的属性路由key
  52.       args.put("x-dead-letter-routing-key", DEAD_LETTER_ROUTING_KEY_A);
  53.       // 声明队列的消息TTL存活时间
  54.       args.put("x-message-ttl", 10000);
  55.       return QueueBuilder.durable(DELAY_QUEUE_NAME_A).withArguments(args).build();
  56.    }
  57.  
  58.    // 声明延迟队列B，延迟60s，并且绑定到对应的死信交换机
  59.    @Bean("delayQueueB")
  60.    public Queue delayQueueB() {
  61.       HashMap<String, Object> args = new HashMap<>();
  62.       // 声明队列绑定的死信交换机
  63.       args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE_NAME);
  64.       // 声明队列的属性路由key
  65.       args.put("x-dead-letter-routing-key", DEAD_LETTER_ROUTING_KEY_B);
  66.       // 声明队列的消息TTL存活时间
  67.       args.put("x-message-ttl", 60000);
  68.       return QueueBuilder.durable(DELAY_QUEUE_NAME_B).withArguments(args).build();
  69.    }
  70.  
  71.    // 声明死信队列A，用于接收延迟10S的消息
  72.    @Bean("deadLetterQueueA")
  73.    public Queue deadLetterQueueA() {
  74.       return new Queue(DEAD_LETTER_QUEUE_NAME_A);
  75.    }
  76.  
  77.    // 声明死信队列B，用于接收延迟60S的消息
  78.    @Bean("deadLetterQueueB")
  79.    public Queue deadLetterQueueB() {
  80.       return new Queue(DEAD_LETTER_QUEUE_NAME_B);
  81.    }
  82.  
  83.    // 设置延迟队列A的绑定关系
  84.    @Bean
  85.    public Binding delayBindingA(@Qualifier("delayQueueA") Queue queue,
  86.                                 @Qualifier("delayExchange") DirectExchange exchange) {
  87.       return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUE_ROUTING_KEY_A);
  88.    }
  89.  
  90.    // 设置延迟队列B的绑定关系
  91.    @Bean
  92.    public Binding delayBindingB(@Qualifier("delayQueueB") Queue queue,
  93.                                 @Qualifier("delayExchange") DirectExchange exchange) {
  94.       return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUE_ROUTING_KEY_B);
  95.    }
  96.  
  97.    // 设置死信队列A的绑定关系
  98.    @Bean
  99.    public Binding deadLetterBindingA(@Qualifier("deadLetterQueueA") Queue queue,
  100.                                 @Qualifier("deadLetterExchange") DirectExchange exchange) {
  101.       return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_ROUTING_KEY_A);
  102.    }
  103.    // 设置死信队列B的绑定关系
  104.    @Bean
  105.    public Binding deadLetterBindingB(@Qualifier("deadLetterQueueB") Queue queue,
  106.                                      @Qualifier("deadLetterExchange") DirectExchange exchange) {
  107.       return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_ROUTING_KEY_B);
  108.    }
  109. }

此配置文件的代码关系图如下

 生产者

  1. import static com.ying.demo.config.RabbitMQConfiguration.DELAY_EXCHANGE_NAME;
  2. import static com.ying.demo.config.RabbitMQConfiguration.DELAY_QUEUE_ROUTING_KEY_A;
  3. import static com.ying.demo.config.RabbitMQConfiguration.DELAY_QUEUE_ROUTING_KEY_B;
  4. /**
  5.  * @author 小影
  6.  * @create: 2022/8/18 11:13
  7.  * @describe：延迟消息生产者
  8.  */
  9. @Component
  10. public class DelayMessageProducer {
  11.  
  12.    @Resource
  13.    private RabbitTemplate rabbitTemplate;
  14.  
  15.    public void send(String message,int type) {
  16.       switch (type){
  17.          case 1: // 10s的消息
  18.             // param：队列名称、路由key、消息
  19.             rabbitTemplate.convertAndSend(DELAY_EXCHANGE_NAME, DELAY_QUEUE_ROUTING_KEY_A, message);
  20.             break;
  21.          case 2:// 60s的消息
  22.             rabbitTemplate.convertAndSend(DELAY_EXCHANGE_NAME, DELAY_QUEUE_ROUTING_KEY_B, message);
  23.             break;
  24.       }
  25.    }
  26. }

消费者

  1. import com.rabbitmq.client.Channel;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springframework.amqp.core.Message;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Component;
  6. import java.time.LocalDateTime;
  7. import static com.ying.demo.config.RabbitMQConfiguration.DEAD_LETTER_QUEUE_NAME_A;
  8. import static com.ying.demo.config.RabbitMQConfiguration.DEAD_LETTER_QUEUE_NAME_B;
  9.  
  10. /**
  11.  * @author 小影
  12.  * @create: 2022/8/18 11:19
  13.  * @describe：死信消费者
  14.  */
  15. @Slf4j
  16. @Component
  17. public class DeadLetterQueueConsumer {
  18.  
  19.    /**
  20.     * 监听私信队列A
  21.     * @param message
  22.     * @param channel 作手动回执、确认
  23.     */
  24.    @RabbitListener(queues = DEAD_LETTER_QUEUE_NAME_A)
  25.    public void receiveA(Message message, Channel channel) {
  26.       String msg = new String(message.getBody());
  27.       log.info("当前时间：{}，死信队列A收到消息：{}", LocalDateTime.now(),msg);
  28.    }
  29.  
  30.    /**
  31.     * 监听私信队列B
  32.     * @param message
  33.     * @param channel 作手动回执、确认
  34.     */
  35.    @RabbitListener(queues = DEAD_LETTER_QUEUE_NAME_B)
  36.    public void receiveB(Message message, Channel channel) {
  37.       String msg = new String(message.getBody());
  38.       log.info("当前时间：{}，死信队列B收到消息：{}", LocalDateTime.now(),msg);
  39.    }
  40. }

测试

  1. @Slf4j
  2. @RestController
  3. @RequestMapping("rabbitmq")
  4. public class RabbitMqController {
  5.    @Resource
  6.    private DelayMessageProducer producer;
  7.  
  8.    @GetMapping("send")
  9.    public void send(String message, Integer type) {
  10.       log.info("当前时间：{}，消息：{}，延迟类型：{}", LocalDateTime.now(), message, Objects.requireNonNull(type));
  11.       producer.send(message, type);
  12.    }
  13. }

分别请求：

http://localhost:8089/rabbitmq/send?message=我是10秒&type=1

http://localhost:8089/rabbitmq/send?message=我是60秒&type=2

如果出现异常：Channel shutdown: channel error; protocol method：#method(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'type' for exchange 'delay.exchange' in vhost '/': received ''x-delayed-message'' but current is 'direct', class-id=40, method-id=10

可能是mq已经存在交换机了先去删掉

弊端：后期要扩展其他不同延时的时间，就需要增加延时的配置，非常麻烦

🏴 写法二 (死信队列TTL)

生产者生产消息(并设置过期时间)——>到交换机分发给延迟队列——>过期后到死信交换机——>消费者进行消费（执行顺序如下图）

 RabbitMQ配置文件

  1. import org.springframework.amqp.core.*;
  2. import org.springframework.beans.factory.annotation.Qualifier;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5.  
  6. import java.util.HashMap;
  7.  
  8. /**
  9.  * @author 小影
  10.  * @create: 2022/8/18 10:26
  11.  * @describe：mq配置 如示例图配置：2交换机、2队列、2路由key
  12.  */
  13. @Configuration
  14. public class RabbitMQConfiguration {
  15.    // 延迟交换机
  16.    public static final String DELAY_EXCHANGE_NAME = "delay.exchange";
  17.    // 延迟队列
  18.    public static final String DELAY_QUEUE_NAME = "delay.queue";
  19.    // 延迟队列路由key
  20.    public static final String DELAY_QUEUE_ROUTING_KEY = "delay.routingKey";
  21.  
  22.    // 死信交换机
  23.    public static final String DEAD_LETTER_EXCHANGE_NAME = "dead.letter.exchange";
  24.    // 死信队列
  25.    public static final String DEAD_LETTER_QUEUE_NAME = "dead.letter.queue";
  26.    // 私信队列路由key
  27.    public static final String DEAD_LETTER_ROUTING_KEY = "dead.letter.routingkey";
  28.  
  29.    // 声明延迟交换机
  30.    @Bean("delayExchange")
  31.    public DirectExchange delayExchange() {
  32.       return new DirectExchange(DELAY_EXCHANGE_NAME);
  33.    }
  34.  
  35.    // 声明死信交换机
  36.    @Bean("deadLetterExchange")
  37.    public DirectExchange deadLetterExchange() {
  38.       return new DirectExchange(DEAD_LETTER_EXCHANGE_NAME);
  39.    }
  40.  
  41.    // 声明延迟队列,不设置存活时间，并且绑定到对应的死信交换机
  42.    @Bean("delayQueue")
  43.    public Queue delayQueue() {
  44.       HashMap<String, Object> args = new HashMap<>();
  45.       // 声明队列绑定的死信交换机
  46.       args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE_NAME);
  47.       // 声明队列的属性路由key
  48.       args.put("x-dead-letter-routing-key", DEAD_LETTER_ROUTING_KEY);
  49.       return QueueBuilder.durable(DELAY_QUEUE_NAME).withArguments(args).build();
  50.    }
  51.  
  52.  
  53.    // 声明死信队列
  54.    @Bean("deadLetterQueue")
  55.    public Queue deadLetterQueue() {
  56.       return new Queue(DEAD_LETTER_QUEUE_NAME);
  57.    }
  58.  
  59.  
  60.    // 设置延迟队列的绑定关系
  61.    @Bean
  62.    public Binding delayBinding(@Qualifier("delayQueue") Queue queue,
  63.                                @Qualifier("delayExchange") DirectExchange exchange) {
  64.       return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUE_ROUTING_KEY);
  65.    }
  66.  
  67.  
  68.    // 设置死信队列的绑定关系
  69.    @Bean
  70.    public Binding deadLetterBinding(@Qualifier("deadLetterQueue") Queue queue,
  71.                                     @Qualifier("deadLetterExchange") DirectExchange exchange) {
  72.       return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_ROUTING_KEY);
  73.    }
  74.  
  75. }

生产者

  1. import static com.ying.demo.config.RabbitMQConfiguration.DELAY_EXCHANGE_NAME;
  2. import static com.ying.demo.config.RabbitMQConfiguration.DELAY_QUEUE_ROUTING_KEY;
  3. /**
  4.  * @author 小影
  5.  * @create: 2022/8/18 11:13
  6.  * @describe：延迟消息生产者
  7.  */
  8. @Component
  9. public class DelayMessageProducer {
  10.  
  11.    @Resource
  12.    private RabbitTemplate rabbitTemplate;
  13.  
  14.    /**
  15.     *
  16.     * @param message 消息
  17.     * @param delayTime 存活时间
  18.     */
  19.    public void send(String message,String delayTime) {
  20.       // param：延迟交换机，路由KEY，存活时间
  21.       rabbitTemplate.convertAndSend(DELAY_EXCHANGE_NAME, DELAY_QUEUE_ROUTING_KEY, message, msg -> {
  22.          msg.getMessageProperties().setExpiration(delayTime);
  23.          return msg;
  24.       });
  25.    }
  26. }

消费者

  1. import com.rabbitmq.client.Channel;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springframework.amqp.core.Message;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Component;
  6. import java.time.LocalDateTime;
  7. import static com.ying.demo.config.RabbitMQConfiguration.DEAD_LETTER_QUEUE_NAME;
  8.  
  9. /**
  10.  * @author 小影
  11.  * @create: 2022/8/18 11:19
  12.  * @describe：死信消费者
  13.  */
  14. @Slf4j
  15. @Component
  16. public class DeadLetterQueueConsumer {
  17.  
  18.    /**
  19.     * 监听私信队列A
  20.     * @param message
  21.     * @param channel 作手动回执、确认
  22.     */
  23.    @RabbitListener(queues = DEAD_LETTER_QUEUE_NAME)
  24.    public void receiveA(Message message, Channel channel) {
  25.       String msg = new String(message.getBody());
  26.       log.info("当前时间：{}，死信队列收到消息：{}", LocalDateTime.now(),msg);
  27.    }
  28.  
  29. }

测试

  1. @Slf4j
  2. @RestController
  3. @RequestMapping("rabbitmq")
  4. public class RabbitMqController {
  5.    @Resource
  6.    private DelayMessageProducer producer;
  7.    @GetMapping("send")
  8.    public void send(String message, String delayTime) {
  9.       log.info("当前时间：{}，消息：{}，存活时间：{}", LocalDateTime.now(), message, delayTime);
  10.       producer.send(message, delayTime);
  11.  
  12.    }
  13. }

分别请求

http://localhost:8089/rabbitmq/send?message=我是60秒&delayTime=60000

http://localhost:8089/rabbitmq/send?message=我是10秒&delayTime=10000

弊端：由于是先进先出的，如果60秒进去了，10秒在进去，10秒结束了，他要等60秒结束，60秒出来10秒才能出来

🚩 写法三 (插件版本-推荐)

安装插件后会生成新的Exchange类型 x-delayed-message ，该类型消息支持延迟投递机制，接收消息后并未立即将消息投递至目标队列，而是存储在mnesia(一个分布式数据库)中，随后检测消息延迟时间，如达到投递时间讲其通过 x-delayed-type 类型标记的交换机投至目标队列。 

插件安装

1.进入mq官网社区插件：

2.找到rabbitmq_delayed_message_exchange

 选择对应版本的ez文件下载

 

 

 注：我的MQ是通过yum安装的

 1.在系统中查看安装的rabbitmq

rpm -qa |grep rabbitmq

 2.查询mq的安装的相关文件目录

rpm -ql rabbitmq-server-3.10.7-1.el8.noarch

 翻到最下面发现mnesia的安装目录； mnesia=分布式数据库，看看就好

 然后把我们下载的ez安装包解压放到 /usr/lib/rabbitmq/lib/rabbitmq_server-3.10.7/plugins 里面

3.重启RabbitMQ服务

systemctl restart rabbitmq-server.service

4.重启插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

 

RabbitMQ配置文件

  1. /**
  2.  * @author 小影
  3.  * @create: 2022/8/18 10:26
  4.  * @describe：mq配置 如示例图配置：1交换机、1队列、1路由key
  5.  */
  6. @Configuration
  7. public class RabbitMQConfiguration {
  8.    // 延迟交换机
  9.    public static final String DELAY_EXCHANGE_NAME = "delay.exchange";
  10.    // 延迟队列
  11.    public static final String DELAY_QUEUE_NAME = "delay.queue";
  12.    // 延迟队列路由key
  13.    public static final String DELAY_QUEUE_ROUTING_KEY = "delay.routingKey";
  14.  
  15.    // 声明延迟交换机
  16.    @Bean("delayExchange")
  17.    public CustomExchange delayExchange() {
  18.       HashMap<String, Object> args = new HashMap<>();
  19.       args.put("x-delayed-type", "direct");
  20.       return new CustomExchange(DELAY_EXCHANGE_NAME,"x-delayed-message",true,false,args);
  21.    }
  22.  
  23.  
  24.    // 声明延迟队列
  25.    @Bean("delayQueue")
  26.    public Queue delayQueue() {
  27.       return new Queue(DELAY_QUEUE_NAME);
  28.    }
  29.  
  30.  
  31.    // 设置延迟队列的绑定关系
  32.    @Bean
  33.    public Binding delayBinding(@Qualifier("delayQueue") Queue queue,
  34.                                @Qualifier("delayExchange") CustomExchange exchange) {
  35.       return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUE_ROUTING_KEY).noargs();
  36.    }
  37. }

生产者

  1. import static com.ying.demo.config.RabbitMQConfiguration.DELAY_EXCHANGE_NAME;
  2. import static com.ying.demo.config.RabbitMQConfiguration.DELAY_QUEUE_ROUTING_KEY;
  3. /**
  4.  * @author 小影
  5.  * @create: 2022/8/18 11:13
  6.  * @describe：延迟消息生产者
  7.  */
  8. @Component
  9. public class DelayMessageProducer {
  10.  
  11.    @Resource
  12.    private RabbitTemplate rabbitTemplate;
  13.  
  14.    /**
  15.     *
  16.     * @param message 消息
  17.     * @param delayTime 存活时间
  18.     */
  19.    public void send(String message,Integer delayTime) {
  20.       // param：延迟交换机，路由KEY，存活时间
  21.       rabbitTemplate.convertAndSend(DELAY_EXCHANGE_NAME, DELAY_QUEUE_ROUTING_KEY, message, msg -> {
  22.          msg.getMessageProperties().setDelay(delayTime);
  23.          return msg;
  24.       });
  25.    }
  26. }

消费者

  1. import static com.ying.demo.config.RabbitMQConfiguration.DELAY_QUEUE_NAME;
  2.  
  3. /**
  4.  * @author 小影
  5.  * @create: 2022/8/18 11:19
  6.  * @describe：消费者
  7.  */
  8. @Slf4j
  9. @Component
  10. public class DeadLetterQueueConsumer {
  11.  
  12.    /*
  13.     * 监听私信队列
  14.     * @param message
  15.     * @param channel 作手动回执、确认
  16.     */
  17.    @RabbitListener(queues = DELAY_QUEUE_NAME)
  18.    public void receiveA(Message message, Channel channel) {
  19.       String msg = new String(message.getBody());
  20.       log.info("当前时间：{}，延迟队列收到消息：{}", LocalDateTime.now(),msg);
  21.    }
  22.  
  23. }

测试

  1. @Slf4j
  2. @RestController
  3. @RequestMapping("rabbitmq")
  4. public class RabbitMqController {
  5.    @Resource
  6.    private DelayMessageProducer producer;
  7.    @GetMapping("send")
  8.    public void send(String message, Integer delayTime) {
  9.       log.info("当前时间：{}，消息：{}，存活时间：{}", LocalDateTime.now(), message, delayTime);
  10.       producer.send(message, delayTime);
  11.  
  12.    }
  13. }

启动项目查看rabbitmq的可视化界面

如下图此时生成的交换机是x-delayed-message类型的

 分别发送：

http://localhost:8089/rabbitmq/send?message=我是60秒&delayTime=60000

http://localhost:8089/rabbitmq/send?message=我是10秒&delayTime=10000

 结局并不是60秒先被消费，完成了我们的意愿。

原理：

  1. 交换机里面有个数据库，生产者生产信息把这个信息放入数据库中
  2. 交换机里面的插件就会一直监听这个时间
  3. 时间到了把对应数据取出来，放入队列，让消费者进行消费

👍 延迟队列方法推荐 

 这是小编在开发学习使用和总结，  这中间或许也存在着不足，希望可以得到大家的理解和建议。如有侵权联系小编！

网站建设定制开发 软件系统开发定制 定制软件开发 软件开发定制 定制app开发 app开发定制 app开发定制公司 电商商城定制开发 定制小程序开发 定制开发小程序 客户管理系统开发定制 定制网站 定制开发 crm开发定制 开发公司 小程序开发定制 定制软件 收款定制开发 企业网站定制开发 定制化开发 android系统定制开发 定制小程序开发费用 定制设计 专注app软件定制开发 软件开发定制定制 知名网站建设定制 软件定制开发供应商 应用系统定制开发 软件系统定制开发 企业管理系统定制开发 系统定制开发