RabbitMQ：伪延时队列

Posted 2020-12-30 kelin-

目录

• 一、什么是延时队列
• 二、RabbitMQ实现
• 三、 延时队列的问题
• 四、解决RabbitMQ的伪延时方案

ps:伪延时队列先卖个关子，我们先了解下延时队列。

一、什么是延时队列

 所谓延时队列是指消息push到队列后，监听的消费者不能第一时间获取消息，需要等到指定时间才能消费。
 一般在业务里面需要对某些消息做定时发送，不想走定时任务或者是用户下单之后多长时间自动失效类似的场景可以考虑通过延时队列实现。

二、RabbitMQ实现

MQ本身并不支持直接的延时队列实现，但是我们可以通过RabbitMQ的消息TTL和Dead Letter规则来实现

1. Time TO Live (TTL):
   RabbitMQ可以针对Queue设置x-expires 或者 针对Message设置 x-message-ttl，来控制消息的生存时间

2. Dead Letter 死信
   RabbitMQ官网这样定义死信消息：

• . 消息被拒绝（basic.reject或basic.nack）并且requeue=false.
• . 消息TTL过期
• 队列达到最大长度（队列满了，无法再添加数据到mq中）

1. Dead Letter Exchanges（DLX）死信交换机
   MQ默认的死信消息是丢弃的，但是我们可以通过设置以下两个属性让死信消息转发到我们指定的队列。

• x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange
• x-dead-letter-routing-key：出现dead letter之后将dead letter重新按照指定的routing-key发送

1. 延时队列实现：
   了解了MQ队列的TTL和Dead Letter之后，我们就可以通过这两个特性来实现，首先我们通过设置消息或者队列的TTL来设置消息在指定时间后成为死信，再设置死信消息的路由转发规则到特定队列，消费者通过监听这个特定队列就能实现延时队列的效果。

2. 代码实现

生产者发送消息：ttlQueue存放过期时间的队列，deadLetterQueue死信转发队列，seconds是过期时间

 public static void sendTTLMsg(String ttlQueue, String deadLetterQueue, Object msg, Integer seconds) {
        MqSender.getInstance().setHost(RABBIT_MQ_HOST);
        // 获取到连接以及MQ通道
        Connection connection;
        try {
            connection = MqSender.getInstance().newConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 配置
            Map<String, Object> args = new HashMap<String, Object>();
            args.put("x-dead-letter-exchange", "");
            args.put("x-dead-letter-routing-key", deadLetterQueue);
            channel.queueDeclare(deadLetterQueue, true, false, false, null);
            channel.queueDeclare(ttlQueue, true, false, false, args);
            // 发送消息
            channel.basicPublish("", ttlQueue, new AMQP.BasicProperties.Builder().expiration(String.valueOf(seconds)).build(), MAPPER.writeValueAsBytes(msg));
            channel.close();
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

消费者通过监听deadLetterQueue来实现延时消息监听

三、 延时队列的问题

通过我们测试发现，这种方式实现的延时队列，在队列设置TTL的情况下是可以正常的，但是如果根据消息设置了不同的TTL，就会有问题，因为MQ本质上还是消息队列中间件，队列是遵循先进先出的，如果有两个消息先后入队，但是后入队的消息TTL小于前面的消息，它必须等待之前的消息被消费完后才能挪到队列头部，这样不同延时消息就会出现问题。

通过RabbitMQ官网的文档也介绍了这个问题：

 Only when expired messages reach the head of a queue will they actually be discarded (or dead-lettered)

所以我才称之为MQ的伪延时队列，这种延时队列在消息TTL不同的情况下并不能实现真正的延时消费。

四、解决RabbitMQ的伪延时方案

既然RabbitMQ无法支持不同TTL消息的延时消费，那么如果我们要实现这种功能，有什么方案呢，在实际业务开发中，我们有这样的解决方案：

首先我们会创建多级延时消费队列（比如两分钟，三十分钟，一天三种，具体可以根据业务量和访问量还有时间精确度来划分，这里的两分钟、三十分钟是指队列统一的TTL），push消费队列的时候，会根据需要延时的时间，丢到不同的消费队列，比如小于三十分钟的我们push到两分钟队列，三十分钟到一天的放入三十分钟队列，超过一天的放入一天队列，在死信队列的监听器做同样的判断，如果是小于等于当前时间消息的，立马消费，否则按照上述规则继续循环到不同的延时队列

这种方案解决了多级延时消费的问题，并且能够较大程度地避免了消息的重复循环，降低MQ的压力，但是缺点也比较明显，因为最低是两分钟的延时，理论上来说最多会有两分钟的误差，如果对时间要求性比较高的，可以适当调低最低一级别的延时TTL，比如一分钟或者三十秒

类似代码如下：cts是需要消费掉的时间戳

 long now = System.currentTimeMillis();
        long cts = Long.valueOf(feedComment.getCts());
        if (cts - now <= 30 * 60 * 1000) {
            MqSender.sendTTLMsg(MqConstants.FEED_COMMENT_DELAY_QUEUE_2MIN, MqConstants.FEED_COMMENT_AUTO_POST_QUEUE, feedComment, 2 * 60);
        } else if (cts - now <= 24 * 60 * 60 * 1000) {
            MqSender.sendTTLMsg(MqConstants.FEED_COMMENT_DELAY_QUEUE_30MIN, MqConstants.FEED_COMMENT_AUTO_POST_QUEUE, feedComment, 30 * 60);
        } else {
            MqSender.sendTTLMsg(MqConstants.FEED_COMMENT_DELAY_QUEUE_24HOUR, MqConstants.FEED_COMMENT_AUTO_POST_QUEUE, feedComment, 24 * 60 * 60);
        }