RabbitMQ高级特性（四）：RabbitMQ之TTL（存活时间/过期时间）
TTL 全称 Time To Live（存活时间/过期时间）。
- 当消息到达存活时间后，还没有被消费，会被自动清除。
- RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。
一、生产者工程
（1）RabbitMQ配置文件（rabbitmq.properties）
rabbitmq.host=192.168.116.161
rabbitmq.port=5672
rabbitmq.username=xiao
rabbitmq.password=xiao
rabbitmq.virtual-host=/myhost
（2）声明ttl队列、交换机，设置队列消息过期时间（spring-rabbitmq-producer.xml）
步骤：
① 声明ttl队列、ttl交换机、交换机绑定队列。
② 在队列参数中设置队列消息过期时间。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
https://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
<!--加载rabbitmq配置文件-->
<context:property-placeholder location="classpath:rabbitmq.properties"/>
<!-- 定义rabbitmq connectionFactory -->
<rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
port="${rabbitmq.port}"
username="${rabbitmq.username}"
password="${rabbitmq.password}"
virtual-host="${rabbitmq.virtual-host}"
publisher-confirms="true"
publisher-returns="true"/>
<!--定义管理交换机、队列-->
<rabbit:admin connection-factory="connectionFactory"/>
<!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
<rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
<!-- ======= 测试消息过期时间TTL ========================================================= -->
<!--（1）声明ttl队列，在队列参数中设置队列过期时间-->
<rabbit:queue name="test_queue_ttl" id="test_queue_ttl">
<rabbit:queue-arguments>
<!--x-message-ttl指队列消息的过期时间，value单位：ms-->
<entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"></entry>
</rabbit:queue-arguments>
</rabbit:queue>
<!--（2）声明ttl交换机（这里使用通配符topic模式），并将ttl队列绑定到该交换机上 -->
<rabbit:topic-exchange name="test_exchange_ttl" >
<rabbit:bindings>
<rabbit:binding pattern="ttl.#" queue="test_queue_ttl"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>
<!-- ================================================================================== -->
</beans>
（3）测试类（为整个队列消息设置TTL）
package net.xiaof.test;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 测试消息TTL（过期时间）
* 1. 队列统一过期
* 2. 消息单独过期
* 如果设置了消息的过期时间，也设置了队列的过期时间，它以时间短的为准。
* 队列过期后，会将队列所有消息全部移除。
* 消息过期后，只有消息在队列顶端，才会判断其是否过期(移除掉)
*/
@Test
public void testTtl() {
for (int i = 1; i <= 10; i++) {
// 发送消息
rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "（"+i+"） message ttl...");
}
}
}
运行结果：
（4）测试类（为单条消息设置TTL）
考虑到单条消息的过期时间与整个队列的过期时间的先后顺序，这里将spring-rabbitmq-producer.xml中整个队列的过期时间增大到100秒：
然后修改测试代码，设置单条消息的过期时间：
package net.xiaof.test;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 测试消息TTL（过期时间）
* 1. 队列统一过期
* 2. 消息单独过期
* 如果设置了消息的过期时间，也设置了队列的过期时间，它以时间短的为准。
* 队列过期后，会将队列所有消息全部移除。
* 消息过期后，只有消息在队列顶端，才会判断其是否过期(移除掉)
*/
@Test
public void testTtl() {
// 消息后处理对象，可以设置一些消息的参数信息
MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//设置消息过期时间（单位：ms）
message.getMessageProperties().setExpiration("5000");//消息的过期时间5秒
return message;
}
};
for (int i = 0; i < 10; i++) {
if (i == 5) {
//消息单独过期
rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....", messagePostProcessor);
} else {
//不过期的消息
rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....");
}
}
}
}
运行结果：
二、TTL总结
- 设置队列过期时间使用参数：
x-message-ttl，单位：ms(毫秒)，会对整个队列消息统一过期。
<!-- ======= 测试消息过期时间TTL ========================================================= -->
<!--（1）声明ttl队列，在队列参数中设置队列过期时间-->
<rabbit:queue name="test_queue_ttl" id="test_queue_ttl">
<rabbit:queue-arguments>
<!--x-message-ttl指队列消息的过期时间，value单位：ms-->
<entry key="x-message-ttl" value="100000" value-type="java.lang.Integer"></entry>
</rabbit:queue-arguments>
</rabbit:queue>
<!--（2）声明ttl交换机（这里使用通配符topic模式），并将ttl队列绑定到该交换机上 -->
<rabbit:topic-exchange name="test_exchange_ttl" >
<rabbit:bindings>
<rabbit:binding pattern="ttl.#" queue="test_queue_ttl"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>
- 设置消息过期时间使用参数：expiration。单位：ms(毫秒)，当该消息在队列头部时（消费时），会单独判断这一消息是否过期。
// 消息后处理对象，可以设置一些消息的参数信息
MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//设置消息过期时间（单位：ms）
message.getMessageProperties().setExpiration("5000");//消息的过期时间5秒
return message;
}
};
如果两者都进行了设置，以时间短的为准。
