4.基于RabbitMQ实现延迟队列

4.1延迟队列定义

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

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

1. 通过 RabbitMQ 本身队列的特性来实现，需要使用 RabbitMQ 的死信交换机（Exchange）和消息的存活时间 TTL（Time To Live）。
2. 在 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主要流程

1. 创建两个交换机（Exchange）和两个队列（Queue）：
   • 原始消息队列：将需要延迟处理的消息发送到原始消息队列。
   • 死信队列：设置一个 TTL（Time-To-Live）过期时间，未在规定时间内处理的消息会成为死信消息，被发送到死信队列。
2. 使用 DLX（死信交换机）关联原始消息队列和死信队列：
   • 在原始消息队列声明时，设置该队列的 x-dead-letter-exchange 参数为死信交换机的名称。
   • 在原始消息队列声明时，设置该队列的 x-dead-letter-routing-key 参数为死信队列的路由键。
3. 消费者监听死信队列进行延迟消息处理：
   • 消费者监听死信队列，一旦有死信消息到达，即可进行相应的延迟消息处理逻辑。
4. 生产者发送延迟消息到原始消息队列：
   • 生产者发送消息到原始消息队列，并设置消息的 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*/@BeanQueue dlxQueue() {return new Queue(DLX_QUEUE_NAME, true, false, false);}/*** 死信交换机* @return*/@BeanDirectExchange dlxExchange() {return new DirectExchange(DLX_EXCHANGE_NAME, true, false);}/*** 绑定死信队列和死信交换机* @return*/@BeanBinding dlxBinding() {return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with(DLX_ROUTING_KEY);}/*** 普通消息队列* @return*/@BeanQueue javaboyQueue() {Map<String, Object> args = new HashMap<>();//设置消息过期时间args.put("x-message-ttl", 1000*10);//设置死信交换机args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);//设置死信 routing_keyargs.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);return new Queue(JAVABOY_QUEUE_NAME, true, false, false, args);}/*** 普通交换机* @return*/@BeanDirectExchange javaboyExchange() {return new DirectExchange(JAVABOY_EXCHANGE_NAME, true, false);}/*** 绑定普通队列和与之对应的交换机* @return*/@BeanBinding 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 {@AutowiredRabbitTemplate rabbitTemplate;@Testvoid 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";@BeanQueue queue() {return new Queue(QUEUE_NAME, true, false, false);}@BeanCustomExchange customExchange() {Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");return new CustomExchange(EXCHANGE_NAME, EXCHANGE_TYPE, true, false,args);}@BeanBinding 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 {@AutowiredRabbitTemplate rabbitTemplate;@Testvoid 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 的特定版本不兼容，需要注意版本匹配。
  • 配置复杂性：使用插件可能需要更复杂的配置和管理，需要更深入的了解插件的工作原理和配置规则。