【RabbitMQ | 第四篇】基于RabbitMQ实现延迟队列

在这里插入图片描述

文章目录

4.基于RabbitMQ实现延迟队列
- 4.1延迟队列定义
- 4.2基于DLX（死信交换机）实现延迟队列
- - 4.2.1实现思路
  - 4.2.2主要流程
  - 4.2.3实战
  - - （1）创建两个消息队列：原始消息队列、死信队列 and 为原始消息队列关联私信交换机
    - （2）为死信队列配置消费者
    - （3）测试
- 4.3基于插件实现延迟队列
- - 4.3.1安装插件
  - 4.3.2创建队列
  - 4.3.3消息消费者
  - 4.3.4发送消息测试
- 4.4两种实现延迟队列的方法优缺点
- - 4.4.1基于死信队列
  - 4.4.2基于插件

4.基于RabbitMQ实现延迟队列

4.1延迟队列定义

延迟队列指的是存储对应的延迟消息，消息被发送以后，并不想让消费者立刻拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费。

RabbitMQ 本身是没有延迟队列的，要实现延迟消息，一般有两种方式：

通过 RabbitMQ 本身队列的特性来实现，需要使用 RabbitMQ 的死信交换机（Exchange）和消息的存活时间 TTL（Time To Live）。
在 RabbitMQ 3.5.7 及以上的版本提供了一个插件（rabbitmq-delayed-message-exchange）来实现延迟队列功能。同时，插件依赖 Erlang/OPT 18.0 及以上。

也就是说，AMQP 协议以及 RabbitMQ 本身没有直接支持延迟队列的功能，但是可以通过 TTL 和 DLX 模拟出延迟队列的功能。

4.2基于DLX（死信交换机）实现延迟队列

4.2.1实现思路

假如一条消息需要延迟 30 分钟执行，我们就设置这条消息的有效期为 30 分钟，同时为这条消息配置死信交换机和死信 routing_key，并且不为这个消息队列设置消费者，那么 30 分钟后，这条消息由于没有被消费者消费而进入死信队列，此时我们有一个消费者就在“蹲点”这个死信队列，消息一进入死信队列，就立马被消费了。

4.2.2主要流程

创建两个交换机（Exchange）和两个队列（Queue）：
- 原始消息队列：将需要延迟处理的消息发送到原始消息队列。
- 死信队列：设置一个 TTL（Time-To-Live）过期时间，未在规定时间内处理的消息会成为死信消息，被发送到死信队列。
使用 DLX（死信交换机）关联原始消息队列和死信队列：
- 在原始消息队列声明时，设置该队列的 x-dead-letter-exchange 参数为死信交换机的名称。
- 在原始消息队列声明时，设置该队列的 x-dead-letter-routing-key 参数为死信队列的路由键。
消费者监听死信队列进行延迟消息处理：
- 消费者监听死信队列，一旦有死信消息到达，即可进行相应的延迟消息处理逻辑。
生产者发送延迟消息到原始消息队列：
- 生产者发送消息到原始消息队列，并设置消息的 TTL（即过期时间）。
- 如果消息在规定时间内未被消费者消费，则成为死信消息。

4.2.3实战

（1）创建两个消息队列：原始消息队列、死信队列 and 为原始消息队列关联私信交换机

@Configuration
public class QueueConfig {
    public static final String JAVABOY_QUEUE_NAME = "javaboy_queue_name";
    public static final String JAVABOY_EXCHANGE_NAME = "javaboy_exchange_name";
    public static final String JAVABOY_ROUTING_KEY = "javaboy_routing_key";
    public static final String DLX_QUEUE_NAME = "dlx_queue_name";
    public static final String DLX_EXCHANGE_NAME = "dlx_exchange_name";
    public static final String DLX_ROUTING_KEY = "dlx_routing_key";

    /**
     * 死信队列
     * @return
     */
    @Bean
    Queue dlxQueue() {
        return new Queue(DLX_QUEUE_NAME, true, false, false);
    }

    /**
     * 死信交换机
     * @return
     */
    @Bean
    DirectExchange dlxExchange() {
        return new DirectExchange(DLX_EXCHANGE_NAME, true, false);
    }

    /**
     * 绑定死信队列和死信交换机
     * @return
     */
    @Bean
    Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange())
                .with(DLX_ROUTING_KEY);
    }

    /**
     * 普通消息队列
     * @return
     */
    @Bean
    Queue javaboyQueue() {
        Map<String, Object> args = new HashMap<>();
        //设置消息过期时间
        args.put("x-message-ttl", 1000*10);
        //设置死信交换机
        args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
        //设置死信 routing_key
        args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);
        return new Queue(JAVABOY_QUEUE_NAME, true, false, false, args);
    }

    /**
     * 普通交换机
     * @return
     */
    @Bean
    DirectExchange javaboyExchange() {
        return new DirectExchange(JAVABOY_EXCHANGE_NAME, true, false);
    }

    /**
     * 绑定普通队列和与之对应的交换机
     * @return
     */
    @Bean
    Binding javaboyBinding() {
        return BindingBuilder.bind(javaboyQueue())
                .to(javaboyExchange())
                .with(JAVABOY_ROUTING_KEY);
    }
}

总结
- 创建两个消息队列：普通消息队列 / 死信队列（配置队列中的消息过期时间时，默认的时间单位时毫秒。）
- 每个队列均分别绑定一个交换器（普通交换机通过 routing key绑定该普通消息队列；死信交换机通过 routing key绑定该私死信交换机）
- 重要：
  - 在原始消息队列声明时，设置该队列的 x-dead-letter-exchange 参数为死信交换机的名称。
  - 在原始消息队列声明时，设置该队列的 x-dead-letter-routing-key 参数为死信队列的路由键。

（2）为死信队列配置消费者

@Component
public class DlxConsumer {
    private static final Logger logger = LoggerFactory.getLogger(DlxConsumer.class);

    @RabbitListener(queues = QueueConfig.DLX_QUEUE_NAME)
    public void handle(String msg) {
        logger.info(msg);
    }
}

（3）测试

@SpringBootTest
class DelayQueueApplicationTests {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() {
        System.out.println(new Date());
        rabbitTemplate.convertAndSend(QueueConfig.JAVABOY_EXCHANGE_NAME, QueueConfig.JAVABOY_ROUTING_KEY, "hello javaboy!");
    }

}

10 秒之后这条消息会在死信队列的消费者中被打印出来。

4.3基于插件实现延迟队列

4.3.1安装插件

首先我们需要下载 rabbitmq_delayed_message_exchange 插件，这是一个 GitHub 上的开源项目，我们直接下载即可：

https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

选择适合自己的版本，我这里选择最新的 3.9.0 版。

下载完成后在命令行执行如下命令将下载文件拷贝到 Docker 容器中去：

docker cp ./rabbitmq_delayed_message_exchange-3.9.0.ez some-rabbit:/plugins

这里第一个参数是宿主机上的文件地址，第二个参数是拷贝到容器的位置。

接下来再执行如下命令进入到 RabbitMQ 容器中：

docker exec -it some-rabbit /bin/bash

进入到容器之后，执行如下命令启用插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

启用成功之后，还可以通过如下命令查看所有安装的插件，看看是否有我们刚刚安装过的插件，如下：

rabbitmq-plugins list

命令的完整执行过程如下图：

OK，配置完成之后，接下来我们执行 exit 命令退出 RabbitMQ 容器。然后开始编码。

4.3.2创建队列

@Configuration
public class RabbitConfig {
    public static final String QUEUE_NAME = "javaboy_delay_queue";
    public static final String EXCHANGE_NAME = "javaboy_delay_exchange";
    public static final String EXCHANGE_TYPE = "x-delayed-message";

    @Bean
    Queue queue() {
        return new Queue(QUEUE_NAME, true, false, false);
    }

    @Bean
    CustomExchange customExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(EXCHANGE_NAME, EXCHANGE_TYPE, true, false,args);
    }
    
    @Bean
    Binding binding() {
        return BindingBuilder.bind(queue())
                .to(customExchange()).with(QUEUE_NAME).noargs();
    }
}

这里主要是交换机的定义有所不同，小伙伴们需要注意。

这里我们使用的交换机是 CustomExchange，这是一个 Spring 中提供的交换机，创建 CustomExchange 时有五个参数，含义分别如下：

交换机名称。
交换机类型，这个地方是固定的。
交换机是否持久化。
如果没有队列绑定到交换机，交换机是否删除。
其他参数。

最后一个 args 参数中，指定了交换机消息分发的类型，这个类型就是大家熟知的 direct、fanout、topic 以及 header 几种，用了哪种类型，将来交换机分发消息就按哪种方式来。

4.3.3消息消费者

@Component
public class MsgReceiver {
    private static final Logger logger = LoggerFactory.getLogger(MsgReceiver.class);
    @RabbitListener(queues = RabbitConfig.QUEUE_NAME)
    public void handleMsg(String msg) {
        logger.info("handleMsg,{}",msg);
    }
}

4.3.4发送消息测试

@SpringBootTest
class MqDelayedMsgDemoApplicationTests {

    @Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    void contextLoads() throws UnsupportedEncodingException {
        Message msg = MessageBuilder.withBody(("hello 江南一点雨"+new Date()).getBytes("UTF-8")).setHeader("x-delay", 3000).build();
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME, RabbitConfig.QUEUE_NAME, msg);
    }

}

4.4两种实现延迟队列的方法优缺点

4.4.1基于死信队列

优点：
- 相对简单，不需要安装额外的插件或组件。
- 可以通过设置消息的 TTL 来实现延迟投递，非常灵活。
可能出现的问题：
- 时间精度限制：TTL 是以毫秒为单位的，可能无法满足某些特定的时间精度要求。
- 大量超时消息：如果系统中积压了大量超时消息，可能会导致死信队列中的消息堆积，增加系统负担。

4.4.2基于插件

优点：
- 提供了更灵活的配置选项，可以根据具体需求定制化设置。
- 可能提供更高的时间精度和可靠性。
可能出现的问题：
- 插件兼容性：某些插件可能与 RabbitMQ 的特定版本不兼容，需要注意版本匹配。
- 配置复杂性：使用插件可能需要更复杂的配置和管理，需要更深入的了解插件的工作原理和配置规则。

在这里插入图片描述