RabbitMQ延迟消息的实现

RabbitMQ延迟队列的实现

延迟消息是什么

延迟消息是将消息发送到MQ中，消费者不会立即收到消息，而是过一段时间之后才会收到消息，进行处理。在一些业务中，可以用到延迟消息，比如我们在成功下单一个商品后，需要立即付款，为了避免商品库存一直被占有，我们会给商品设置一个支付时间，如果在这段时间没有支付成功，就会恢复库存，删除订单，对于订单支付的超时删除我们是通过延迟消息来实现的，让消费者在支付超时之后查询用户是否支付，如果支付成功直接返回，如果支付失败就恢复库存删除订单。

延迟消息的实现

延迟消息由以下两种方式实现，第一种是通过绑定死信交换机实现，第二种通过延迟消息插件实现，推荐使用第二种，更加简单

死信交换机

满足以下三种情况之一的叫做死信：
1、在设置了过期时间的消息，放入队列中，超过了过期时间没有被处理的消息
2、消息消费失败（返回nack或者reject）并且不能重复入队
3、队列消息堆积满了，最早的消息叫做死信
我们可以给队列绑定参数指定交换机，那么死信会被投递到指定交换机。
消息队列实现原理：我们可以设置一组没有消费者的交换机和队列，设置另一组处理绑定死信的交换机、队列和消费者，来处理延迟消息。

代码实现

定义死信交换机等和消费延迟消息交换机等：

@Configuration
public class DelayConfiguration {
    /**
     * 定义死信交换机、队列以及绑定
     */
    @Bean
    public DirectExchange exchange() {
        return new DirectExchange("dead.direct");
    }
    @Bean
    public Queue queue() {
        Queue queue = new Queue("dead.queue");
        queue.addArgument("x-dead-letter-exchange", "delay.direct");
        return queue;
    }
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with("dead");
    }
    /**
     * 定义处理延迟消息的交换机、队列和绑定
     */
    @Bean
    public DirectExchange exchange1() {
        return new DirectExchange("delay.direct");
    }
    @Bean
    public Queue queue1() {
        return new Queue("delay.queue");
    }
    @Bean
    public Binding binding1() {
        return BindingBuilder.bind(queue1()).to(exchange1()).with("dead");
    }
}

定义延迟消息监听器：

    @RabbitListener(queues = "delay.queue")
    public void listen(String msg){
        log.info(LocalDateTime.now()+": "+msg);
    }

测试：

@Test
    void sendDeadMsg() {
        rabbitTemplate.convertAndSend("dead.direct", "dead", "我是死信", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
//                设置过期消息时间
                message.getMessageProperties().setExpiration("10000");
                return message;
            }
        });
    }

结果：


消费者在十秒钟后成功消费延迟消息

延迟消息插件

我们在之前通过死信交换机来实现延迟队列，但是死信交换机是专门用来存放无法处理的消息，并且使用死信交换机实现过于复杂，我们需要手动定义两个交换机和队列，因而RabbitMQ提供了延迟消息插件来让我们更简单的实现延迟消息。
原理：给消息设置延迟时间，当将消息放入MQ时，MQ的交换机不会立即将消息放入队列，而是会在交换机中暂存延迟时间过后将消息路由到队列中，可以让队列处理延迟消息。
安装插件：插件安装可以借鉴这篇博客
代码实现：
消费者：

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "delay.queue",durable = "true"),
            // 开启延迟交换机
            exchange = @Exchange(name = "delay.direct",delayed = "true"),
            key = "dead"
    ))
    public void listen(String msg){
        log.info(msg);
    }

  @Test
    void sendDeadMsg() {
        rabbitTemplate.convertAndSend("delay.direct", "delay", "我是死信", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
//                设置延迟消息时间
                message.getMessageProperties().setDelay(10000);
                return message;
            }
        });
    }