RabbitMQ：延时队列

快来打我* 2022-05-18 09:46 207阅读 0赞

> # **RabbitMQ：延时队列** #

### **（一）：在队列上设置TTL ** ###

[参考文献：RabbitMQ学习之延时队列][RabbitMQ]

![48590-20170706094322534-766388494.png][]

** 其中，【延时队列delay.1m.queue】配置参数如下需配置：**

![70][]

**【生产者配置】：**

package com.caox.rabbitmq.demo._15_rabbitmq_delay_query_ttl;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    
    /**
     * Created by nazi on 2018/8/1.
     * 延时队列 队列配置延时
     */
    
    
    public class Producer {
    
        private static String queue_name = "test.queue";
    
        public static void main(String[] args) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setUsername("caoxia");
            factory.setPassword("caoxia123456");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(queue_name, true, false, false, null);
            String message = "hello world!" + System.currentTimeMillis();
            channel.basicPublish("delaysync.exchange", "deal.message", null, message.getBytes());
            System.out.println("sent message: " + message + ",date:" + System.currentTimeMillis());
            // 关闭频道和连接
            channel.close();
            connection.close();
        }
    }

**【消费者配置】：**

package com.caox.rabbitmq.demo._15_rabbitmq_delay_query_ttl;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.QueueingConsumer;
    
    /**
     * Created by nazi on 2018/8/1.
     * 延时队列 队列配置延时
     */
    
    public class Consumer {
    
        private static String queue_name = "test.queue";
    
        public static void main(String[] args) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setUsername("caoxia");
            factory.setPassword("caoxia123456");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(queue_name, true, false, false, null);
            // 绑定路由
            channel.queueBind(queue_name, "delay.exchange", "deal.message");
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 指定消费队列
            channel.basicConsume(queue_name, true, consumer);
            while (true) {
                // nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println("received message:" + message + ",date:" + System.currentTimeMillis());
            }
        }
    
    }

### **（二）：在消息上设置TTL  ** ###

![Center][]

**【生产者配置】：**

package com.caox.rabbitmq.demo._14_rabbitmq_delay_news_ttl;
    
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.util.HashMap;
    /**
     * Created by nazi on 2018/8/1.
     * 延时队列 消息配置延时
     */
    
    public class ProducerUpt {
    
        private static String queue_name = "message_ttl_queue";
    
        public static void main(String[] args) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setUsername("caoxia");
            factory.setPassword("caoxia123456");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            HashMap<String, Object> arguments = new HashMap<String, Object>();
            arguments.put("x-dead-letter-exchange", "amq.direct");
            arguments.put("x-dead-letter-routing-key", "message_ttl_routingKey");
            channel.queueDeclare("delay_queue", true, false, false, arguments);
    
            // 声明队列
            channel.queueDeclare(queue_name, true, false, false, null);
            // 绑定路由
            channel.queueBind(queue_name, "amq.direct", "message_ttl_routingKey");
    
            String message = "hello world!" + System.currentTimeMillis();
            // 设置延时属性
            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
            // 持久性 non-persistent (1) or persistent (2)
            AMQP.BasicProperties properties = builder.expiration("120000").deliveryMode(2).build();
            // routingKey =delay_queue 进行转发
            channel.basicPublish("", "delay_queue", properties, message.getBytes());
            System.out.println("sent message: " + message + ",date:" + System.currentTimeMillis());
            // 关闭频道和连接
            channel.close();
            connection.close();
        }
    }

**【消费者配置】： **

package com.caox.rabbitmq.demo._14_rabbitmq_delay_news_ttl;
    
    import java.util.HashMap;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.QueueingConsumer;
    
    /**
     * Created by nazi on 2018/8/1.
     * 延时队列 消息配置延时
     */
    
    
    public class ConsumerUpt {
    
        private static String queue_name = "message_ttl_queue";
    
        public static void main(String[] args) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setUsername("caoxia");
            factory.setPassword("caoxia123456");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            HashMap<String, Object> arguments = new HashMap<String, Object>();
            arguments.put("x-dead-letter-exchange", "amq.direct");
            arguments.put("x-dead-letter-routing-key", "message_ttl_routingKey");
            channel.queueDeclare("delay_queue", true, false, false, arguments);
    
            // 声明队列
            channel.queueDeclare(queue_name, true, false, false, null);
            // 绑定路由
            channel.queueBind(queue_name, "amq.direct", "message_ttl_routingKey");
    
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 指定消费队列
            channel.basicConsume(queue_name, true, consumer);
            while (true) {
                // nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println("received message:" + message + ",date:" + System.currentTimeMillis());
            }
        }
    
    }

**【结论】：**

**虽然 consumer 从来看不到过期的 message ，但是在过期 message 到达 queue 的头部时确实会被真正的丢弃（或者 dead-lettered ）。**

**【在队列上设置TTL】：当对每一个 queue 设置了 TTL 值时不会产生任何问题，因为过期的 message 总是会出现在 queue 的头部。一旦消息过期，就会从队列中抹去。**

**【在消息上设置TTL】：****当对每一条 message 设置了 TTL 时，过期的 message 可能会排队于未过期 message 的后面，直到这些消息被 consume 到或者过期了。在这种情况下，这些过期的 message 使用的资源将不会被释放，且会在 queue 统计信息中被计算进去（例如，queue 中存在的 message 的数量）。即使消息过期，也不会马上从队列中抹去，因为每条消息是否过期时在即将投递到消费者之前判定的。为什么两者得处理方法不一致？因为第一种方法里，队列中已过期的消息肯定在队列头部，RabbitMQ只要定期从队头开始扫描是否有过期消息即可，而第二种方法里，每条消息的过期时间不同，如果要删除所有过期消息，势必要扫描整个队列，所以不如等到此消息即将被消费时再判定是否过期，如果过期，再进行删除。**

[RabbitMQ]: https://blog.csdn.net/zhu_tianwei/article/details/53563311
[48590-20170706094322534-766388494.png]: /images/20220518/384010f0ad114c3c9efdc79c771a0ee1.png
[70]: /images/20220518/f02419f399334e92b57e49e0f432e1ca.png
[Center]: /images/20220518/c12976531f2c4313acf590637300e376.png