0
0

RabbitMQ延迟队列

💌 介绍

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

💒 使用场景

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

以上场景都可以用延时队列来完成

🏳‍🌈 模拟案例

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

📕 准备工作

导入RabbitMQ依赖

  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
  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;
  6. import java.util.HashMap;
  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";
  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";
  33.    // 声明延迟交换机
  34.    @Bean("delayExchange")
  35.    public DirectExchange delayExchange() {
  36.       return new DirectExchange(DELAY_EXCHANGE_NAME);
  37.    }
  39.    // 声明死信交换机
  40.    @Bean("deadLetterExchange")
  41.    public DirectExchange deadLetterExchange() {
  42.       return new DirectExchange(DEAD_LETTER_EXCHANGE_NAME);
  43.    }
  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.    }
  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.    }
  71.    // 声明死信队列A，用于接收延迟10S的消息
  72.    @Bean("deadLetterQueueA")
  73.    public Queue deadLetterQueueA() {
  74.       return new Queue(DEAD_LETTER_QUEUE_NAME_A);
  75.    }
  77.    // 声明死信队列B，用于接收延迟60S的消息
  78.    @Bean("deadLetterQueueB")
  79.    public Queue deadLetterQueueB() {
  80.       return new Queue(DEAD_LETTER_QUEUE_NAME_B);
  81.    }
  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.    }
  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.    }
  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 {
  12.    @Resource
  13.    private RabbitTemplate rabbitTemplate;
  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;
  10. /**
  11.  * @author 小影
  12.  * @create: 2022/8/18 11:19
  13.  * @describe：死信消费者
  14.  */
  15. @Slf4j
  16. @Component
  17. public class DeadLetterQueueConsumer {
  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.    }
  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;
  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;
  6. import java.util.HashMap;
  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";
  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";
  29.    // 声明延迟交换机
  30.    @Bean("delayExchange")
  31.    public DirectExchange delayExchange() {
  32.       return new DirectExchange(DELAY_EXCHANGE_NAME);
  33.    }
  35.    // 声明死信交换机
  36.    @Bean("deadLetterExchange")
  37.    public DirectExchange deadLetterExchange() {
  38.       return new DirectExchange(DEAD_LETTER_EXCHANGE_NAME);
  39.    }
  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.    }
  52.    // 声明死信队列
  53.    @Bean("deadLetterQueue")
  54.    public Queue deadLetterQueue() {
  55.       return new Queue(DEAD_LETTER_QUEUE_NAME);
  56.    }
  58.    // 设置延迟队列的绑定关系
  59.    @Bean
  60.    public Binding delayBinding(@Qualifier("delayQueue") Queue queue,
  61.                                @Qualifier("delayExchange") DirectExchange exchange) {
  62.       return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUE_ROUTING_KEY);
  63.    }
  65.    // 设置死信队列的绑定关系
  66.    @Bean
  67.    public Binding deadLetterBinding(@Qualifier("deadLetterQueue") Queue queue,
  68.                                     @Qualifier("deadLetterExchange") DirectExchange exchange) {
  69.       return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_ROUTING_KEY);
  70.    }
  72. }

生产者

  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 {
  11.    @Resource
  12.    private RabbitTemplate rabbitTemplate;
  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;
  9. /**
  10.  * @author 小影
  11.  * @create: 2022/8/18 11:19
  12.  * @describe：死信消费者
  13.  */
  14. @Slf4j
  15. @Component
  16. public class DeadLetterQueueConsumer {
  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.    }
  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);
  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官网社区插件：Community Plugins — RabbitMQ

2.找到rabbitmq_delayed_message_exchange

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

Releases · rabbitmq/rabbitmq-delayed-message-exchange · GitHub

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

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

  1. rpm -qa |grep rabbitmq

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

  1. 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服务

  1. systemctl restart rabbitmq-server.service

4.重启插件

  1. 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";
  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.    }
  23.    // 声明延迟队列
  24.    @Bean("delayQueue")
  25.    public Queue delayQueue() {
  26.       return new Queue(DELAY_QUEUE_NAME);
  27.    }
  29.    // 设置延迟队列的绑定关系
  30.    @Bean
  31.    public Binding delayBinding(@Qualifier("delayQueue") Queue queue,
  32.                                @Qualifier("delayExchange") CustomExchange exchange) {
  33.       return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUE_ROUTING_KEY).noargs();
  34.    }
  35. }

生产者

  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 {
  11.    @Resource
  12.    private RabbitTemplate rabbitTemplate;
  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;
  3. /**
  4.  * @author 小影
  5.  * @create: 2022/8/18 11:19
  6.  * @describe：消费者
  7.  */
  8. @Slf4j
  9. @Component
  10. public class DeadLetterQueueConsumer {
  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.    }
  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);
  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. 时间到了把对应数据取出来，放入队列，让消费者进行消费

👍 延迟队列方法推荐

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

标签： rabbitmq 分布式 java
本文转载自: https://blog.csdn.net/weixin_46522803/article/details/126521534
版权归原作者 小影~ 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ延迟队列”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？