RabbitMQ延时队列的使用

r囧r小猫 2023-10-11 14:58 16阅读 0赞

### 延时队列概述 ###

通过给一个普通的队列设置一些参数，死信交换机，死信路由，以及队列过期时间，注意，这个队列不能让任何消费者监听，然后我们将消息统一发往这个队列，当这个队列中的消息过期后，就会根据我们指定的死信交换机和死信路由去往新的队列，我们的程序可以监听这个队列，一旦这个队列监听到消息了，则一定是延时队列中过期的消息。

*  流程图如下

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2_size_16_color_FFFFFF_t_70_pic_center]

### 延时队列测试使用 ###

*  引入AMQP的依赖

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

*  编写一个配置类，给容器中注入一些bean，当容器启动时，会自定根据这些bean去rabbit服务器创建响应队列与交换机。

@Slf4j
    @Configuration
    public class RabbitConfig {
        
    
        @Bean
        public MessageConverter messageConverter(){
        
            return new Jackson2JsonMessageConverter();
        }
    
        /**
         * 创建延时队列
         */
        @Bean
        public Queue orderDelayQueue(){
        
            Map<String,Object> map = new HashMap<>();
            //消息过期后交给这个交换机处理
            map.put("x-dead-letter-exchange","order-event-exchange");
            //消息过期后交给上面的交换机，该交换机再根据这个路由键转发给那个处理消息的队列
            map.put("x-dead-letter-routing-key","order.release.order");
            //设置队列的过期时间
            map.put("x-message-ttl",60000);
            Queue queue = new Queue("order.delay.queue",true,false,
                    false,map);
            return queue;
        }
    
        /**
         * 当延时队列中的消息过期时将消息发往这个队列
         */
        @Bean
        public Queue orderReleaseOrderQueue(){
        
            Queue queue = new Queue("order.release.order.queue",true,false,
                    false,null);
            return queue;
        }
    
        /**
         * 交换机,这个交换机同时绑定延时队列和订单释放队列，需要根据不同的routing-key来转发到不同的队列
         * 所以我们选择用Topic类型交换机
         */
        @Bean
        public Exchange orderEventExchange(){
        
            TopicExchange exchange = new TopicExchange(
                    "order-event-exchange",
                    true,
                    false,
                    null
            )    ;
            return exchange;
        }
    
        /**
         * 将交换机和队列进行绑定，同时可以绑定两个队列
         */
        @Bean
        public Binding orderCreateOrder(){
        
            return new Binding(
                    "order.delay.queue",
                    Binding.DestinationType.QUEUE,
                    "order-event-exchange",
                    "order.create.order", //延时队列绑定的路由键，创建订单时根据这个路由键发送到延时队列
                    null
            );
        }
    
        @Bean
        public Binding orderReleaseOrder(){
        
            return new Binding(
                    "order.release.order.queue",
                    Binding.DestinationType.QUEUE,
                    "order-event-exchange",
                    "order.release.order", //消费队列绑定的路由键，延时队列中的消息失效后根据这个路由键发送给响应的队列
                    null
            );
        }
    
        /**
         * 测试监听队列,监听的是延时队列过期后的处理队列
         */
        @RabbitListener(queues = "order.release.order.queue")
        public void listener(Message message, OrderDTO orderDTO, Channel channel){
        
            log.info("收到过期的订单信息，准备关闭订单，{}",orderDTO);
        }
    }

*  其中我们在创建延时队列时，传递了一个Map，这个map就是指定了我们延时队列的一些属性信息。

登录rabbit服务器查看，order.delay.queue这个队列确实多了这三个属性。  
![在这里插入图片描述][20201105092800407.png_pic_center]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20201105092058539.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2,size_16,color_FFFFFF,t_70#pic_center
[20201105092800407.png_pic_center]: https://img-blog.csdnimg.cn/20201105092800407.png#pic_center