消息中间件 - RabbitMQ延迟队列

雨点打透心脏的1/2处 2022-09-08 00:20 286阅读 0赞

## 一、概念 ##

延时队列的内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素是希望在指定时间到了以后或之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

## 二、应用场景 ##

1.  订单在十分钟之内未支付则自动取消。
2.  新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
3.  用户注册成功后，如果三天内没有登陆则进行短信提醒。
4.  用户发起退款，如果三天内没有得到处理则通知相关运营人员。
5.  预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议

这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务，如：发生订单生成事件，在十分钟之后检查该订单支付状态，然后将未支付的订单进行关闭；看起来似乎使用定时任务，一直轮询数据，每秒查一次，取出需要被处理的数据，然后处理不就完事了吗？如果数据量比较少，确实可以这样做，比如：对于“如果账单一周内未支付则进行自动结算”这样的需求，如果对于时间不是严格限制，而是宽松意义上的一周，那么每天晚上跑个定时任务检查一下所有未支付的账单，确实也是一个可行的方案。但对于数据量比较大，并且时效性较强的场景，如：“订单十分钟内未支付则关闭“，短期内未支付的订单数据可能会有很多，活动期间甚至会达到百万甚至千万级别，对这么庞大的数据量仍旧使用轮询的方式显然是不可取的，很可能在一秒内无法完成所有订单的检查，同时会给数据库带来很大压力，无法满足业务要求而且性能低下。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBATXJESnVu_size_29_color_FFFFFF_t_70_g_se_x_16]

## 三、RabbitMQ中的TTL ##

TTL 是什么呢？ TTL 是 RabbitMQ 中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间 ，单位是毫秒。换句话说，如果一条消息设置了 TTL 属性或者进入了设置 TTL 属性的队列，那么这条消息如果在 TTL 设置的时间内没有被消费，则会成为"死信"。如果同时配置了队列的 TTL 和消息的TTL，那么较小的那个值将会被使用，有两种方式设置 TTL 。

### （1）消息设置TTL ###

针对每条消息设置 TTL，消息过期后可以投递到死信队列或者直接丢弃。

// 配置消息的过期时间为10秒
    AMQP.BasicProperties properties = new AMQP.BasicProperties()
        .builder()
        .expiration("10000")
        .build();
    channel.basicPublish("exchange", "routing_key", properties, (msg).getBytes(StandardCharsets.UTF_8));

### （2）队列设置TTL ###

创建队列的时候设置队列的 “x-message-ttl” 属性 。设置了TTL属性的队列，消息过期后投递到死信队列或直接丢弃，还有一种处理方式，消息即使过期，也不一定会被马上丢弃 ，因为**消息是否过期是在即将投递到消费者之前判定的**，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长的时间。

如果不设置 TTL，表示消息永远不会过期，如果将 TTL 设置为 0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

## 四、初步实现延时队列 ##

生产者P将消息发送给交换机X，X根据路由键XA将消息投递到队列QA中等待10秒，或者根据路由键XB投递消息到QB队列等待40s，延时超时后，根据路由键YD投递到死信交换机Y，再转发到队列QD，最后由消费者C消费队列QD的消息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBATXJESnVu_size_35_color_FFFFFF_t_70_g_se_x_16]  
结合SpringBoot 做一个代码实现：

通过一个配置文件，注入各个交换机和各个队列，在绑定交换机与队列之间的关系。

import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    public class TTLQueueConfig { 
        /** * 延时队列-示例使用常量 */
        String X_EXCHANGE = "xExchange";
        /** 队列A，延时 10s 后，消息转发给死信交换机B，再转发给队列C */
        String DELAY_QUEUE_A_NAME = "qa";
        /** 队列B，延时 20s 后，消息转发给死信交换机B，再转发给队列C */
        String DELAY_QUEUE_B_NAME = "qb";
        /** 死信队列交换机 */
        String Y_DEAD_LETTER_EXCHANGE = "yExchange";
        String DEAD_LETTER_QUEUE = "qd";
        /** 交换机x投递的路由键 */
        String ROUTING_KEY_DELAY_XA = "xa";
        /** 交换机y投递的路由键 */
        String ROUTING_KEY_DELAY_XB = "xb";
        /** 由死信交换机Y投递到死信队列 yd */
        String ROUTING_KEY_DELAY_YD = "yd";
        
        /** * 注册交换机X */
        @Bean(X_EXCHANGE)
        public DirectExchange xExchange() { 
            return new DirectExchange(X_EXCHANGE);
        }
    
        /** * 注册交换机Y（死信） */
        @Bean(Y_DEAD_LETTER_EXCHANGE)
        public DirectExchange yExchange() { 
            return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
        }
    
        /** * 声明队列 A ttl 为 10s 并绑定到对应的死信交换机 */
        @Bean(DELAY_QUEUE_A_NAME)
        public Queue queueA() { 
            Map<String, Object> args = new HashMap<>(4);
            // 声明当前队列绑定的死信交换机
            args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
            // 声明当前队列的死信路由 key
            args.put("x-dead-letter-routing-key", ROUTING_KEY_DELAY_YD);
            // 声明队列的 TTL
            args.put("x-message-ttl", 10 * 1000);
            return QueueBuilder.durable(DELAY_QUEUE_A_NAME).withArguments(args).build();
        }
    
        /** * 声明队列 A 绑定 X 交换机 */
        @Bean
        public Binding queueABindingX(@Qualifier(DELAY_QUEUE_A_NAME) Queue queueA,
                                      @Qualifier(X_EXCHANGE) DirectExchange xExchange) { 
            return BindingBuilder.bind(queueA).to(xExchange).with(ROUTING_KEY_DELAY_XA);
        }
    
        /** * 声明队列 A ttl 为 20s 并绑定到对应的死信交换机 */
        @Bean(DELAY_QUEUE_B_NAME)
        public Queue queueB() { 
            Map<String, Object> args = new HashMap<>(4);
            // 声明当前队列绑定的死信交换机
            args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
            // 声明当前队列的死信路由 key
            args.put("x-dead-letter-routing-key", ROUTING_KEY_DELAY_YD);
            // 声明队列的 TTL
            args.put("x-message-ttl", 20 * 1000);
            return QueueBuilder.durable(DELAY_QUEUE_B_NAME).withArguments(args).build();
        }
    
        /** * 声明队列 B 绑定 X 交换机 */
        @Bean
        public Binding queueBBindingX(@Qualifier(DELAY_QUEUE_B_NAME) Queue queueB,
                                      @Qualifier(X_EXCHANGE) DirectExchange xExchange) { 
            return BindingBuilder.bind(queueB).to(xExchange).with(ROUTING_KEY_DELAY_XB);
        }
    
        /** * 死信队列 D */
        @Bean(DEAD_LETTER_QUEUE)
        public Queue queueD() { 
            return new Queue(DEAD_LETTER_QUEUE);
        }
    
        /** * 死信队列 D 绑定交换机Y */
        @Bean
        public Binding queueDBindingY(@Qualifier(DEAD_LETTER_QUEUE) Queue queueD,
                                      @Qualifier(Y_DEAD_LETTER_EXCHANGE) DirectExchange yExchange) { 
            return BindingBuilder.bind(queueD).to(yExchange).with(ROUTING_KEY_DELAY_YD);
        }
    }

这里方便测试，以Controller作为生产者发送消息。

@Slf4j
    @RestController
    @RequestMapping("/ttl")
    public class TTLMessageDelayQueueProducer { 
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @GetMapping("/sendMsg/{message}")
        public void sendMsg(@PathVariable String message) { 
            log.info("当前时间： {},发送一条信息给两个 TTL 队列:{}", new Date(), message);
            rabbitTemplate.convertAndSend(X_EXCHANGE, ROUTING_KEY_DELAY_XA, "消息来自 ttl 为 10S 的队列: " + message);
            rabbitTemplate.convertAndSend(X_EXCHANGE, ROUTING_KEY_DELAY_XB, "消息来自 ttl 为 20S 的队列: " + message);
        }
    }

消费者只负责消费死信队列D。

import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    
    @Slf4j
    @Component
    public class TTLMessageDelayQueueConsumer { 
        /** 消费死信队列 D */
        @RabbitListener(queues = DEAD_LETTER_QUEUE)
        public void receiveD(Message message, Channel channel) { 
            log.info("当前时间： {},收到死信队列信息{}", new Date(), new String(message.getBody()));
        }
    }

**总结**

第一条消息在 10S 后变成了死信消息，然后被消费者消费掉，第二条消息在 20S 之后变成了死信消息，然后被消费掉，这样一个延时队列就打造完成了。

不过，如果这样使用的话，岂不是每增加一个新的时间需求，就要新增一个队列，这里只提供了10s和20s两个时间选项，如果某些消息有特定的需求延时7天后再处理，那么又需要新增一个延时时长为7天的队列。如果是提前预约医生之后再通知的一个场景，岂不是要增加无数个队列才能满足需求？

## 五、延时队列的优化 ##

在这里新增了一个队列 QC，绑定关系如下，该队列不设置 TTL 时间 。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBATXJESnVu_size_35_color_FFFFFF_t_70_g_se_x_16 1]

### （1）代码实现 ###

TTLQueueConfig 配置文件新增队列C的配置。

public static final String QUEUE_C = "qc";
    public static final String ROUTING_KEY_DELAY_XC = "xc";
    
    /** 声明队列 C 死信交换机 */
    @Bean(name=QUEUE_C)
    public Queue queueC() { 
        Map<String, Object> args = new HashMap<>(3);
        //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //声明当前队列的死信路由 key
        args.put("x-dead-letter-routing-key", ROUTING_KEY_DELAY_YD);
        //没有声明 TTL 属性
        return QueueBuilder.durable(QUEUE_C).withArguments(args).build();
    }
    
    /** 声明队列 B 绑定 X 交换机 */
    @Bean
    public Binding queuecBindingX(@Qualifier(QUEUE_C) Queue queueC,
                                  @Qualifier(X_EXCHANGE) DirectExchange xExchange){ 
        return BindingBuilder.bind(queueC).to(xExchange).with(ROUTING_KEY_DELAY_XC);
    }

消息生产者

@GetMapping("sendExpirationMsg/{message}/{ttlTime}")
    public void sendMsg(@PathVariable String message,@PathVariable String ttlTime) { 
        rabbitTemplate.convertAndSend(X_EXCHANGE, ROUTING_KEY_DELAY_XC, message, correlationData ->{ 
            correlationData.getMessageProperties().setExpiration(ttlTime);
            return correlationData;
        });
        log.info("当前时间： {},发送一条时长{}毫秒 TTL 信息给队列 C:{}", new Date(),ttlTime, message);
    }

### （2）问题分析 ###

消费者发送的消息，先发一条20s的，在发一条2s的，按理应该时间短的先出来，那么结果是怎么样的还需要看消费者。

2021-06-25 13:30:58.634  INFO 13720 --- [nio-8005-exec-8] c.f.t.r.p.b.TTLMessageDelayQueueProducer : 当前时间：Fri Jun 25 13:30:58 CST 2021,发送一条时长：20000毫秒，TTL 信息给队列：测试消息
    2021-06-25 13:31:02.499  INFO 13720 --- [io-8005-exec-10] c.f.t.r.p.b.TTLMessageDelayQueueProducer : 当前时间：Fri Jun 25 13:31:02 CST 2021,发送一条时长：2000毫秒，TTL 信息给队列：测试消息

消费者日志，两条消息都是在13:31:18这个时间出来的，说明并没有让时间短的消息优先出来。

2021-06-25 13:31:18.636  INFO 12724 --- [ntContainer] c.f.t.r.c.b.TTLMessageDelayQueueConsumer : 当前时间：Fri Jun 25 13:31:18 CST 2021,收到队列：qd的信息：测试消息
    2021-06-25 13:31:18.636  INFO 12724 --- [ntContainer] c.f.t.r.c.b.TTLMessageDelayQueueConsumer : 当前时间：Fri Jun 25 13:31:18 CST 2021,收到队列：qd的信息：测试消息

看起来似乎没什么问题，但是在最开始的时候，就介绍过如果使用在消息属性上设置 TTL 的方式，消息可能并不会按时“死亡“，因为 RabbitMQ 只会检查第一个消息是否过期，如果过期则丢到死信队列，**如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行**。要解决这个问题，还得依赖RabbitMQ官方提供的一个插件。

## 六、RabbitMQ 插件实现延时队列 ##

### （1）安装插件 ###

[点击此处前往下载][Link 1] rabbitmq\_delayed\_message\_exchange 插件。使用 docker 安装 RabbitMQ 的延时队列插件。

# 这是使用的是3.8.x版本的RabbitMQ，故安装3.8的插件，拷贝到容器内docker cp /home/soft/rabbitmq_delayed_message_exchange-3.8.0.ez rabbitmq:/plugins# 进入RabbitMQ容器docker exec -it rabbitmq /bin/bash# 查看插件列表rabbitmq-plugins list# 启用插件rabbitmq-plugins enable rabbitmq_delayed_message_exchange

再查看一下插件列表是否启用：

[E*] rabbitmq_delayed_message_exchange 3.8.0

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBATXJESnVu_size_31_color_FFFFFF_t_70_g_se_x_16]

在自定义的交换机中，这是一种新的交换类型，该类型消息支持延迟投递机制 消息传递后并不会立即投递到目标队列中，而是存储在 mnesia(一个分布式数据系统)表中，当达到投递时间时，才投递到目标队列中 。

### （2）插件使用示例 ###

消费者

public class MessageDelayQueueProducer { 
        static final String EXCHANGE_NAME = "delayExchange";
        static final String ROUTING_KEY = "delayRouting";
    
        public static void main(String[] args) throws Exception { 
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.19.128");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
    
            // 声明x-delayed-type类型的exchange
            Map<String, Object> arguments = new HashMap<>(4);
            arguments.put("x-delayed-type", "direct");
            // 持久化
            boolean durable = true;
            channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", durable, false, arguments);
    
            // 设置延迟10s发送消息
            Map<String, Object> headers = new HashMap<>(4);
            headers.put("x-delay", 10 * 1000);
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .headers(headers)
                    .build();
    
            for (int i = 0; i < 10; i++) { 
                String msg = "message_" + i;
                channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, properties, (msg).getBytes(StandardCharsets.UTF_8));
                System.out.println("消息[" + msg + "]发送完成");
            }
    
            channel.close();
            connection.close();
        }
    }

消费者

public class MessageDelayQueueConsumer { 
        static final String QUEUE_NAME = "delay";
        static final String EXCHANGE_NAME = "delayExchange";
        static final String ROUTING_KEY = "delayRouting";
    
        public static void main(String[] args) throws Exception { 
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.19.128");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
    
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
    
            System.out.println("等待接收消息...");
            channel.basicConsume(QUEUE_NAME, true, (consumerTag, delivery) -> { 
                byte[] msgBytes = delivery.getBody();
                System.out.println("接收到消息：消息ID=" + consumerTag + "===>" + new String(msgBytes));
            }, consumerTag -> { 
                System.out.println("消息被中断：消息ID=" + consumerTag);
            });
        }
    }

### （3）结合SpringBoot插件使用示例 ###

在这里新增了一个队列 delayed.queue，一个自定义交换机 delayed.exchange，绑定关系如下：  
![在这里插入图片描述][ca6bd99c84c24cba9b05e62c00505409.png]  
在TTLQueueConfig中新增绑定关系

String DELAY_PLUGIN_QUEUE_NAME = "delayed.queue";
    String DELAY_PLUGIN_EXCHANGE_NAME = "delayed.exchange";
    String ROUTING_KEY_PLUGIN_DELAY = "delay.routing";
    
    @Bean(DELAY_PLUGIN_QUEUE_NAME)
    public Queue delayedQueue() { 
        return new Queue(DELAY_PLUGIN_QUEUE_NAME);
    }
    
    @Bean(DELAY_PLUGIN_EXCHANGE_NAME)
    public CustomExchange delayedExchange() { 
        Map<String, Object> args = new HashMap<>();
        // 自定义交换机的类型
        args.put("x-delayed-type", "direct");
        // 在此可以不设置x-delay属性，可以在发送消息时给每一条消息设置自定义时长
        return new CustomExchange(DELAY_PLUGIN_EXCHANGE_NAME, "x-delayed-message", true, false, args);
    }
    
    @Bean
    public Binding bindingDelayedQueue(@Qualifier(DELAY_PLUGIN_QUEUE_NAME) Queue queue,
                                       @Qualifier(DELAY_PLUGIN_EXCHANGE_NAME) CustomExchange delayedExchange) { 
        return BindingBuilder.bind(queue).to(delayedExchange).with(ROUTING_KEY_PLUGIN_DELAY).noargs();
    }

生产者

/** * 使用RabbitMQ 延时队列插件，自定义消息过期时间 * 解决问题：如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行 */
    @GetMapping("sendDelayMsg/{message}/{delayTime}")
    public void sendMsg(@PathVariable String message, @PathVariable Integer delayTime) { 
        rabbitTemplate.convertAndSend(DELAY_PLUGIN_EXCHANGE_NAME, ROUTING_KEY_PLUGIN_DELAY, message,
                correlationData -> { 
                    correlationData.getMessageProperties().setDelay(delayTime);
                    return correlationData;
                });
        log.info("当前时间：{},发送一条时长：{}毫秒，TTL 信息给队列：{}", new Date(), delayTime, message);
    }

消费者

@RabbitListener(queues = DELAY_PLUGIN_QUEUE_NAME)
    public void receiveDelayed(Message message, Channel channel) { 
        log.info("当前时间：{},收到队列：{}的信息：{}", new Date(), DELAY_PLUGIN_QUEUE_NAME, new String(message.getBody()));
    }

### （4）问题解决 ###

生产者日志

2021-06-25 13:35:47.578  INFO 13720 --- [nio-8005-exec-4] c.f.t.r.p.b.TTLMessageDelayQueueProducer : 当前时间：Fri Jun 25 13:35:47 CST 2021,发送一条时长：20000毫秒，TTL 信息给队列：测试消息
    2021-06-25 13:35:49.043  INFO 13720 --- [nio-8005-exec-5] c.f.t.r.p.b.TTLMessageDelayQueueProducer : 当前时间：Fri Jun 25 13:35:49 CST 2021,发送一条时长：2000毫秒，TTL 信息给队列：测试消息

消费者日志，再次查看消息出来的时间，发现确实是时间短的消息优先出来。

2021-06-25 13:35:51.029  INFO 12724 --- [ntContainer] c.f.t.r.c.b.TTLMessageDelayQueueConsumer : 当前时间：Fri Jun 25 13:35:51 CST 2021,收到队列：delayed.queue的信息：测试消息
    2021-06-25 13:36:07.389  INFO 12724 --- [ntContainer] c.f.t.r.c.b.TTLMessageDelayQueueConsumer : 当前时间：Fri Jun 25 13:36:07 CST 2021,收到队列：delayed.queue的信息：测试消息

## 七、总结 ##

延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的利用RabbitMQ 的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过 RabbitMQ 集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。

当然，延时队列还有很多其它选择，比如利用 Java 的 DelayQueue，利用 Redis 的 zset，利用 Quartz 或者利用 kafka 的时间轮，这些方式各有特点，看需要适用的场景。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBATXJESnVu_size_29_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/90a4c70f90524fd2affa2bee54e2d35a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBATXJESnVu_size_35_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/595c833b6aff423482afd895b411fb95.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBATXJESnVu_size_35_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/2de5b0aded56477696c2bd0b4db3c999.png
[Link 1]: https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/tag/v3.8.0
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_Q1NETiBATXJESnVu_size_31_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/6c3180466b0246c3825aa887461cf853.png
[ca6bd99c84c24cba9b05e62c00505409.png]: /images/20220829/822f1e8243dd40f181bfd1105d6f484d.png