利用RabbitMQ的TTL实现延迟队列

喜欢ヅ旅行 2023-01-21 11:20 38阅读 0赞

## 利用RabbitMQ的TTL实现延迟队列 ##

`RabbitMQ`自带不支持延迟队列，但是可以利用TTL过期的特性，建立两个队列。生产端将信息发送到有过期时间的队列，当消息长时间不消费(达到过期时间)就会转为私信队列，只需要在过期前将信息压入正常的队列中，让消费端监听这个正常队列就行。

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

首先配置两个队列，一个为死信队列`Queue1`，一个是正常消费的队列`Queue2`

/**
     * 配置两个队列，队列1为死信队列，队列2为正常消费队列
     */
    @Configuration
    public class MqConfig {
        
        /**
         * 死信队列 orderDeadQueue
         * @return orderDeadQueue
         */
        @Bean
        public Queue orderDeadQueue(){
        
            return QueueBuilder.durable("orderDeadQueue")
                    .withArgument("x-dead-letter-exchange","orderExchange")//过期时发送到orderExchange交换机
                    .withArgument("x-dead-letter-routing-key","orderListQueue")//对应orderListQueue队列
                    .build();
    
        }
    
        /**
         * 用户正常监听的队列
         * @return
         */
        @Bean
        public Queue orderListQueue(){
        
            return new Queue("orderListQueue",true);
        }
    
        /**
         * 创建交换机
         */
        @Bean
        public Exchange orderExchange(){
        
            return new DirectExchange("orderExchange");
        }
        /**
         * 绑定交换机和队列
         */
        @Bean
        public Binding binding(@Autowired  Exchange orderExchange,@Autowired  Queue orderListQueue){
        
            return BindingBuilder.bind(orderListQueue).to(orderExchange).with("orderListQueue").noargs();
        }
    }

创建一个监听器用于监听正常消费的队列 `orderListQueue`

@Component
    @RabbitListener(queues = "orderListQueue")
    public class MQDeadListener {
        
        @RabbitHandler
        public void ExecuteMessage(String message){
        
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            System.out.println("消费的时间为："+simpleDateFormat.format(new Date()));
            System.out.println("消费的消息为："+message);
        }
    }

测试：

@SpringBootTest(classes = OrderApplication.class)
    @RunWith(SpringRunner.class)
    public class RabbitMqTest {
        
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        /**
         * 使用rabbitTemplate发送信息到队列1
         * 过期时间为10s
         * 当队列1的消息过期时就会将消息发送到队列2
         * 消费者只需要监听队列2即可
         * 从而实现延迟队列
         */
        @Test
        public void  test01(){
        
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String date = "我觉得还好"+simpleDateFormat.format(new Date());
    
            rabbitTemplate.convertAndSend("orderDeadQueue", (Object) date, new MessagePostProcessor() {
        
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
        
                    message.getMessageProperties().setExpiration(10000+"");
                    return message;
                }
            });
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dxMjMyMw_size_16_color_FFFFFF_t_70]: /images/20221020/81404a1a2e7b48908e366a7400e93508.png