【从零开始学习RabbitMQ | 第三篇】什么是延迟消息

news2024/11/18 10:34:12

前言：

延迟消息：

延迟消息实现方式：

死信交换机：

延迟消息插件：

1.基于注解的方式

2.基于@Bean的方式

总结：

前言：

在现代软件开发中，异步消息处理已成为构建可扩展、高可用系统的关键组成部分。RabbitMQ，作为一款广泛使用的开源消息代理，提供了强大的消息队列功能，支持多种消息模式，包括发布/订阅、请求/响应以及路由等。然而，除了这些基本功能外，RabbitMQ还提供了一项独特的特性——延迟消息，它允许开发者安排消息在将来的某个时间点被处理。

延迟消息，顾名思义，是指那些在发送后不会立即被消费者接收的消息，而是根据预设的延迟时间后才能被消费。这种特性在许多场景下非常有用，例如定时任务的执行、订单超时处理、批处理作业的调度等。通过延迟消息，我们可以将即时处理的需求转化为按计划执行的任务，从而优化资源使用，提高系统的响应性和吞吐量。

延迟消息：

生产者发送消息的时候指定一个时间，消费者不会立即收到消息，而是在指定的时间之后才收到消息。

我举一个实际应用场景：买车票

订单不代表付款，当我们下单之后，只有付款了我们才会更新库存状态 。在苍穹外卖中，我们采用的是定时任务轮询数据库机制，来取消超时订单。

但是这有一个弊端：对数据库的压力太大了，无论是每隔几秒都是对数据库的一次极致拷打。

在黑马点评中，我们用Redis模拟实现了延迟队列，其实就是用Stream去构造订单，过期之后就去数据库查询订单状态。

而在这种环境下，我们可以用RabbitMQ中的延迟队列，当下单之后，我们把消息发送给MQ，设置延迟时间为30分钟，等30分钟后，Rabbitmq检查订单状态，如果未支付就取消订单。

延迟消息实现方式：

死信交换机：

当队列中的消息满足以下一种情况的时候，就会成为死信：

消费者使用basic.reject 或 basic.nack声明消费失败，并且消息的requeue参数设置为fasle
消息是一个过期消息，超时无人消费
要投递的消息队列满了，最早的消息成为死信

当队列通过dead-letter-exchange 属性指定交换机之后，这个队列中的死信就会投递到这个交换机中，这个交换机成为死信交换机。

这种机制的缺点就是代码实现比较麻烦。所以我们接下来介绍一下延迟消息插件。

延迟消息插件：

延迟消息插件是Rabbitmq官方推出的一个插件，原生支持延迟消息，原理是设计了一个支持延迟消息功能的交换机，当消息投递到交换机的时候可以暂存一段时间，到期之后再投递到队列。

Releases · rabbitmq/rabbitmq-delayed-message-exchange (github.com)https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases在上述网址中下载插件之后，存放到RabbitMQ的插件目录下就可以了。

安装延迟消息插件之后，想要实现延迟消息可以通过两桶途径：

1.基于注解的方式

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "delay.queue", durable = "true"),
            exchange = @Exchange(value = "delay.direct", delayed = "true"),
            key = "delay"
    ))
    public void listenDelayQueue(String msg) {
        log.info("delay.queue：" + msg);
    }

2.基于@Bean的方式

@Configuration
public class DirectConfiguration {
 
    @Bean
    public DirectExchange delayExchange() {
        return ExchangeBuilder
                .directExchange("delay.direct")
                .delayed()
                .durable(true)
                .build();
    }
 
    @Bean
    public Queue delayedQueue() {
        return new Queue("delay.queue");
    }
 
    @Bean
    public Binding delayQueueBinding() {
        return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");
    }
}