RabbitMQ学习笔记 - 过期时间（TTL）

墨蓝 2022-01-22 03:13 229阅读 0赞

**参考：<<RabbitMQ实战指南>>**

TTL（Time To Live），过期时间。RabbitMQ可以对消息和队列设置TTL。

## 一、设置消息的TTL ##

设置消息TTL的两种方式：

*  方式一：通过队列属性设置，队列中所有消息都有相同的过期时间
 *  方式二：对消息本身进行单独设置，每条消息的TTL可以不同

注：

*  如果两种方式一起使用，则消息的TTL以两者之间较小的那个数值为准。
 *  消息在队列中的生存时间一旦超过设置的TTL值时，就会变成**死信（Dead Message）**，消费者将无法再收到该消息。（如果队列设置了死信队列，那么这条消息就会被转发到死信队列上，后面可以正常消费）
 *  方式一设置TTL的方式，消息一旦过期，就会从队列中抹去；第二种方式，即使消息过期，也不会马上从队列抹去，而是在消息被投递到消费者之前判定的。

### 1. 通过队列属性设置消息的TTL ###

在声明队列时时，添加**x-message-ttl**参数实现，也就是调用`channel.queueDeclare()`方法时设置，单位：毫秒。

示例：

#### 1.1 原生api ####

// ...
    String queueName = "direct.queue.test.queue.ttl";
    Map<String, Object> arguments = new HashMap<>();
    arguments.put("x-message-ttl", 5000);
    channel.queueDeclare(queueName, durable, exclusive, autoDelete, arguments);
    // ...

注：

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

#### 1.2 springboot ####

（1）声明交换器、队列

和原生api类似，也是在声明队列时设置**x-message-ttl**参数来实现的。

@Bean
    public DirectExchange msgQueueTtlExchange() { 
        return new DirectExchange(msgQueueTtlExchange);
    }
    
    @Bean
    public Queue msgQueue() { 
        // 设置队列消息过期时间
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-message-ttl", 5000);
        return new Queue(msgQueue, true, false, false, arguments);
    }
    
    @Bean
    Binding bindingMsgQueue(Queue msgQueue, DirectExchange msgQueueTtlExchange) { 
        return BindingBuilder.bind(msgQueue).to(msgQueueTtlExchange).with(msgQueueRoutingKey);
    }

（2）测试发送消息

发送消息并没有什么不同，直接发送即可。

@Autowired
    private RabbitTemplate rabbitTemplate;
    
    public boolean sendQueueMsgTTL() { 
        String message = "test queue msg ttl message";
        this.rabbitTemplate.convertAndSend(msgQueueTtlExchange, msgQueueRoutingKey, message);
        return true;
    }

### 2. 设置每条消息的TTL ###

在发送消息时，设置**expiration**属性，也就是`channel.basicPublish()`方法，单位：毫秒。

示例：

#### 2.1 原生api ####

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
            .deliveryMode(2)
            .expiration("5000")// 单位ms
            .build();
    channel.basicPublish(exchangeName, routingKey, false, properties, "Test Msg".getBytes());

#### 2.2 springboot ####

正常声明交换器、队列，不需要设置其他参数，主要在发送消息时，设置这条消息的TTL。

RabbitMQTemplate提供的发送消息api，其他都是这个发送方法的缺少参数的重载方法：

public void convertAndSend(String exchange, String routingKey, final Object message, final MessagePostProcessor messagePostProcessor, CorrelationData correlationData) throws AmqpException

这里我们主要用到`MessagePostProcessor`这个接口，通过实现`Message postProcessMessage(Message message) throws AmqpException`方法，在消息已经经过转换器处理之后，再进一步修改消息，主要修改消息的Header（消息头）和Property（属性）。

发送消息示例：

String content = "test ttl message";
    this.rabbitTemplate.convertAndSend(exchange, routingKey, content, message -> { 
        MessageProperties messageProperties = message.getMessageProperties();
        // 设置这条消息的过期时间
        messageProperties.setExpiration("5000");
        return message;
    });

## 二、设置队列的TTL ##

在声明队列时，也就是调用`channel.queueDeclare()`方法，可以通过**x-expires**参数控制队列被自动删除前处于未使用状态的时间，单位：毫秒。

注：

*  该参数不能设置为0，设置为1000，则表示该队列如果1秒钟之内未使用则会被删除。
 *  未使用的意思是：队列上没有任何消费者，队列也没有被重新声明，并且在过期时间段内也未调用过Basic.Get方法。即使不断发送消息，也不算作这个队列在使用中，到期后仍然会被删除。
 *  RabbitMQ 会确保在过期时间到达后将队列删除，但是不能保证删除及时删除。RabbitMQ重启后，持久化的队列的过期时间会被重新计算。

示例：

#### 1. 原生api ####

// ...
    Map<String, Object> arguments = new HashMap<>();
    arguments.put("x-expires", 1800000);// 30min
    channel.queueDeclare(queueName, durable, exclusive, autoDelete, arguments);
    // ...

#### 2. springboot ####

和原生api类似，在声明队列时通过设置**x-expires**参数来实现。  
（1）声明交换器、队列

@Bean
    public DirectExchange queueTtlExchange() { 
        return new DirectExchange(queueTtlExchange);
    }
    
    @Bean
    public Queue queue() { 
        // 设置队列过期时间
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-expires", 5000);
        return new Queue(queue, true, false, false, arguments);
    }
    
    @Bean
    Binding bindingQueue(Queue queue, DirectExchange queueTtlExchange) { 
        return BindingBuilder.bind(queue).to(queueTtlExchange).with(queueRoutingKey);
    }

（2）发送、消费消息

发送消息、消费消息不需要设置特别，正常使用即可。

springboot-rabbitmq-demo测试代码：[https://github.com/mytt-10566/springboot-rabbitmq-demo][https_github.com_mytt-10566_springboot-rabbitmq-demo]

[https_github.com_mytt-10566_springboot-rabbitmq-demo]: https://github.com/mytt-10566/springboot-rabbitmq-demo