死信队列 DLX + TTL 方式实现延迟队列

  1. 队列，意味着内部的元素是有序的，元素的出队和入队是有方向性的，元素从一端进入，从另一端取出
  2. 延时，这是最重要的特性，普通队列中的元素总是等着希望被早点取出处理，而延时队列中的元素则是希望等待特定时间后，消费者才能拿到这个消息进行消费。

    TTL

    TTL 是 RabbitMQ 中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间，单位是毫秒。换句话说，如果一条消息设置了 TTL 属性或者进入了设置 TTL 属性的队列，那么这条消息如果在 TTL 设置的时间内没有被消费，则会成为“死信”。如果同时配置了队列的 TTL 和消息的 TTL，那么较小的那个值将会被使用。
    也就是说我们可以利用这个机制，让消息过期后，变成死信，就又交给我们的死信交换机来处理了~
    延时队列可以解决很多特定场景下，带时间属性的任务需求，如：订单创建半小时内未支付进行取消订单。。。

有两种方式设置 TTL 值，

第一种是在创建队列的时候设置队列的 “x-message-ttl” 属性

  1. @Bean
  2. public Queue delayQueue() {
  3.     String queueName = "delay_queue";
  4.     Map<String, Object> args = new HashMap<>(1);
  5.     args.put("x-message-ttl", "6000");
  6.     return new Queue(queueName, true, false, false, args);
  7. }

另一种方式是针对每条消息设置 TTL

  1. rabbitTemplate.convertAndSend(exchange, routingKey, (message) -> {
  2.  message.getMessageProperties().setExpiration("6000");
  3.  return message;
  4. });

区别：

  • 设置了队列的 TTL 属性，那么一旦消息过期，就会被队列丢弃
  • 给消息设置 TTL 属性，消息过期也不一定会马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果队列存在消息积压问题，那么已过期的消息可能还会存活较长些时间

死信队列 + 消息TTL = 延迟队列

一.设置队列 TTL 属性来实现延迟队列

消息大致流向
死信队列 DLX TTL 方式实现延迟队列 - 图1
延迟队列 (delay_queu_A) 设置 TTL 能让信息在延迟多久后成为死信，成为死信后的消息都会被投递到死信队列中，这样只需要消费者一直消费死信队列(dlx_queue_A) 里就好了，因为里面的消息都是希望被处理的延迟后的消息。

声明交换机、队列以及他们的关系：

  1. // 配置延迟队列
  2. @Bean
  3. public TopicExchange delayExchange() {
  4.     String exchangeName = "delay_exchange";
  5.     return new TopicExchange(exchangeName);
  6. }
  7. @Bean
  8. public Queue delayQueueA() {
  9.     String queueName = "delay_queue_A";
  10.     // 设置死信发送至 dlx_exchange 交换机，设置路由键为 bind.dlx.A
  11.     String dlxExchangeName = "dlx_exchange";
  12.     String bindDlxRoutingKeyA = "bind.dlx.A";
  14.     Map<String, Object> args = new HashMap<>(3);
  15.     // 设置队列的延迟属性，6秒
  16.     args.put("x-message-ttl", 6000);
  17.     args.put("x-dead-letter-exchange", dlxExchangeName);
  18.     args.put("x-dead-letter-routing-key", bindDlxRoutingKeyA);
  19.     return new Queue(queueName, true, false, false, args);
  20. }
  21. @Bean
  22. public Binding bindingDelayExchange() {
  23.     String routingKey = "bind.delay.A";
  24.     return BindingBuilder.bind(delayQueueA()).to(delayExchange()).with(routingKey);
  25. }
  27. // 配置死信队列
  28. @Bean
  29. public TopicExchange dlxExchange() {
  30.     String exchangeName = "dlx_exchange";
  31.     return new TopicExchange(exchangeName);
  32. }
  33. @Bean
  34. public Queue dlxQueueA() {
  35.     String queueName = "dlx_queue_A";
  36.     return new Queue(queueName);
  37. }
  38. @Bean
  39. public Binding bindingDlxExchange() {
  40.     String routingKey = "#.A";
  41.     return BindingBuilder.bind(dlxQueueA()).to(dlxExchange()).with(routingKey);
  42. }

yml

  1. spring:
  2.   rabbitmq:
  3.     host: 192.168.159.129
  4.     port: 5672
  5.     username: admin
  6.     password: admin
  7.     # 虚拟host 可以不设置,使用 server 默认 host
  8.     virtual-host:
  9.     listener:
  10.       simple:
  11.         default-requeue-rejected:
  12.         acknowledge-mode: manual

消费者

  1. @RabbitListener(queues = "dlx_queue_A")
  2. public void receiver2(@Payload String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
  3.  System.out.println("当前时间：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + "死信队列 - dlx_queue_A 收到消息：" + msg);
  4.  channel.basicAck(deliveryTag, false);
  5. }

生产者

  1. @Test
  2. public void demo_10_Producer() {
  3.     String exchange = "delay_exchange_A";
  4.     String routingKey = "delay.routing.key.A";
  5.     String msg = "发送给延迟队列 delay_queue_A 的消息";
  6.     System.out.println("当前时间：" + simpleDateFormat.format(new Date()) + "开始发送消息：" + msg);
  7.     rabbitTemplate.convertAndSend(exchange, routingKey, msg);
  8. }

输出

  1. 当前时间：2020-09-27 21:10:20 开始发送消息：发送给延迟队列 delay_queue_A 的消息
  3. 当前时间：2020-09-27 21:10:26 死信队列 - dlx_queue_A 收到消息：发送给延迟队列 delay_queue_A 的消息

缺陷

如果这样使用的话，每增加一个新的时间需求，就要新增一个队列。如需要一个小时后处理，那么就需要增加 TTL 为一个小时的队列，如果此时消息的过期时间不确定或者消息过期时间维度过多，在消费端我们就要去监听多个消息队列，岂不是要增加无数个队列才能满足需求？？

二.设置消息的 TTL 属性来实现延迟队列

设计成一个通用的延时队列，我们可以给不同的消息设置不同的 TTL 过期时间，以达到动态设置延迟时间。

声明交换机、队列以及它们的关系：

  1. // 配置延迟队列
  2. @Bean
  3. public TopicExchange delayExchange() {
  4.     String exchangeName = "delay_exchange";
  5.     return new TopicExchange(exchangeName);
  6. }
  7. @Bean
  8. public Queue delayQueueB() {
  9.     String queueName = "delay_queue_B";
  10.     // 设置死信发送至 dlx_exchange 交换机，设置路由键为 bind_dlx_B
  11.     String dlxExchangeName = "dlx_exchange";
  12.     String bindDlxRoutingKeyB = "bind.dlx.B";
  14.     Map<String, Object> args = new HashMap<>(2);
  15.     args.put("x-dead-letter-exchange", dlxExchangeName);
  16.     args.put("x-dead-letter-routing-key", bindDlxRoutingKeyB);
  17.     return new Queue(queueName, true, false, false, args);
  18. }
  19. @Bean
  20. public Binding bindingDelayExchange() {
  21.     String routingKey = "bind.delay.B";
  22.     return BindingBuilder.bind(delayQueueB()).to(delayExchange()).with(routingKey);
  23. }
  25. // 配置死信队列
  26. @Bean
  27. public TopicExchange dlxExchange() {
  28.     String exchangeName = "dlx_exchange";
  29.     return new TopicExchange(exchangeName);
  30. }
  31. @Bean
  32. public Queue dlxQueueB() {
  33.     String queueName = "dlx_queue_B";
  34.     return new Queue(queueName);
  35. }
  36. @Bean
  37. public Binding bindingDlxExchange() {
  38.     String routingKey = "#.B";
  39.     return BindingBuilder.bind(dlxQueueB()).to(dlxExchange()).with(routingKey);
  40. }

消费者

  1. @RabbitListener(queues = "dlx_queue_B")
  2. public void receiverB(@Payload String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
  3.     System.out.println("当前时间：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 死信队列 - dlx_queue_B 收到消息：" + msg);
  4.     channel.basicAck(deliveryTag, false);
  5. }

生产者

  1. @Test
  2. public void producer_B() throws Exception {
  3.     SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  4.     String exchange = "delay_exchange";
  5.     String routingKey = "bind.delay.B";
  6.     String msg = "我是第一条消息";
  7.     // 延迟时间
  8.     String delayTime = "6000";
  10.     System.out.println("当前时间：" + simpleDateFormat.format(new Date()) + "开始发送消息：" + msg + "  延迟的时间为：" + delayTime);
  11.     rabbitTemplate.convertAndSend(exchange, routingKey, msg, new MyMessagePostProcessor(delayTime));
  13.     Thread.sleep(30000L);
  14. }

MyMessagePostProcessor

  1. /**
  2.  * 因为要给消息设置 TTL，这里创建了一个 MessagePostProcessor 的实例来设置过期时间
  3.  */
  4. class MyMessagePostProcessor implements MessagePostProcessor {
  5.     // 延迟时间 毫秒
  6.     private String delayTime;
  8.     MyMessagePostProcessor(String delayTime) {
  9.         this.delayTime = delayTime;
  10.     }
  12.     @Override
  13.     public Message postProcessMessage(Message message) throws AmqpException {
  14.         // 设置延迟时间
  15.         message.getMessageProperties().setExpiration(delayTime);
  16.         return message;
  17.     }
  18. }

这里的 convertAndSend 使用的第四参数为 MessagePostProcessor，我这里采用构造函数的方式来动态设置消息的过期时间。

另外一种实现方式

  1. package com.matrix.queue.rabbitmq.service.comm;
  3. import com.alibaba.fastjson.JSON;
  4. import lombok.AllArgsConstructor;
  5. import org.springframework.amqp.AmqpException;
  6. import org.springframework.amqp.core.Message;
  7. import org.springframework.amqp.core.MessagePostProcessor;
  8. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  9. import org.springframework.stereotype.Service;
  11. /**
  12.  * @className: CommSendService
  13.  * @package: com.matrix.rabbitmq.service
  14.  * @description: 消息队列通用发送类
  15.  * @datetime: 2021/11/30   18:50
  16.  * @author: swordmeng8@163.com
  17.  */
  18. @Service
  19. @AllArgsConstructor
  20. public class CommSendService {
  22.     /** 使用RabbitTemplate,这提供了接收/发送等等方法 */
  23.     private final RabbitTemplate rabbitTemplate;
  26.     public void commSend(Object obj, long times,String exchange,String exchangeRoutingKey){
  27.         MessagePostProcessor processor = new MessagePostProcessor(){
  28.             @Override
  29.             public Message postProcessMessage(Message message) throws AmqpException {
  30.                 message.getMessageProperties().setExpiration(String.valueOf(times));
  31.                 return message;
  32.             }
  33.         };
  34.         rabbitTemplate.convertAndSend(
  35.             exchange,
  36.             exchangeRoutingKey,
  37.             JSON.toJSONString(obj), processor);
  38.     }
  41. }

效果

  1. 当前时间：2020-09-27 21:20:20 开始发送消息：我是第一条消息  延迟的时间为：6000
  3. 当前时间：2020-09-27 21:20:26 死信队列 - dlx_queue_B 收到消息：我是第一条消息

缺陷

但是上面我们也提到了消息过期也不一定会马上丢弃。消息到了过期时间可能并不会按时“死亡“，因为 RabbitMQ 只会检查第一个消息是否过期，如果过期则丢到死信队列，索引如果第一个消息的延时时长很长，而第二个消息的延时时长很短，则第二个消息并不会优先得到执行。

例子：

生产者

  1. @Test
  2. public void producer_B() throws Exception {
  3.     SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  4.     String exchange = "delay_exchange";
  5.     String routingKey = "bind.delay.B";
  6.     String msg = "我是第一条消息";
  7.     // 延迟时间
  8.     String delayTime = "6000";
  10.     System.out.println("当前时间：" + simpleDateFormat.format(new Date()) + "开始发送消息：" + msg + "  延迟的时间为：" + delayTime);
  11.     rabbitTemplate.convertAndSend(exchange, routingKey, msg, new MyMessagePostProcessor(delayTime));
  13.     msg = "我是第二条消息";
  14.     // 修改延迟时间
  15.     delayTime = "3000";
  17.     System.out.println("当前时间：" + simpleDateFormat.format(new Date()) + "开始发送消息：" + msg + "  延迟的时间为：" + delayTime);
  18.     rabbitTemplate.convertAndSend(exchange, routingKey, msg, new MyMessagePostProcessor(delayTime));
  20.     Thread.sleep(30000L);
  21. }

效果

  1. 当前时间：2020-09-27 21:23:20 开始发送消息：我是第一条消息  延迟的时间为：6000
  2. 当前时间：2020-09-27 21:23:20 开始发送消息：我是第二条消息  延迟的时间为：3000
  4. 当前时间：2020-09-27 21:23:26 死信队列 - dlx_queue_B 收到消息：我是第一条消息
  5. 当前时间：2020-09-27 21:23:26 死信队列 - dlx_queue_B 收到消息：我是第二条消息

可以看到但延迟最久的第一条信息消费后，紧跟其后的已经过期了的第二条消息也接着消费了

总结

延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的利用 RabbitMQ 的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过RabbitMQ集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失
死信队列 DLX + TTL 的方式来实现延迟队列，这也是一种通用的做法。
不管哪种方式都有各自的优缺点，根据业务情况来考虑。如果要实现在消息粒度上添加TTL，并使其在设置的TTL时间及时死亡，可以使用 RabbitMQ 的 rabbitmq_delayed_message_exchange插件的方式实现。