1. 过期时间TTL概述
过期时间TTL表示可以对消息设置预期的时间，在这个时间内都可以被消费者接收获取；过了之后消息将自动被删除。RabbitMQ可以对消息和队列设置TTL。
目前有两种方法可以设置。
- 第一种方法是通过队列属性设置，队列中所有消息都有相同的过期时间。
- 第二种方法是对消息进行单独设置，每条消息TTL可以不同。
如果上述两种方法同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值，就称为dead message被投递到死信队列， 消费者将无法再收到该消息。
2.TTL设置方法
2.1. 通过队列属性设置
package com.zju.service.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @author guozonghao
* @Description
* @create 2024-10-04 09:37
*/
@Configuration
public class TTLRabbitMqConfiguration {
@Bean
public DirectExchange ttlDirectExchange(){
return new DirectExchange("ttl_direct_exchange",true,false);
}
@Bean
public Queue directTTLQueue(){
//设置过期时间
HashMap<String, Object> args = new HashMap<>();
args.put("x-message-ttl",5000);//这里一定是int类型
return new Queue("ttl.direct.queue",true,false,false,args);
}
@Bean
public Binding directTTLBinding(){
return BindingBuilder.bind(directTTLQueue()).to(ttlDirectExchange()).with("ttl");
}
}
2.2. 通过消息属性设置
通过AMQP.BasicProperties这个类在发送消息时配置相关的属性
public class MessageTTLProducer {
public static void main(String[] args) {
// 1: 创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
// 2: 设置连接属性
connectionFactory.setHost("47.104.141.27");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("admin");
connectionFactory.setPassword("admin");
Connection connection = null;
Channel channel = null;
try {
// 3: 从连接工厂中获取连接
connection = connectionFactory.newConnection("生产者");
// 4: 从连接中获取通道channel
channel = connection.createChannel();
// 5: 申明队列queue存储消息
/*
* 如果队列不存在，则会创建
* Rabbitmq不允许创建两个相同的队列名称，否则会报错。
*
* @params1： queue 队列的名称
* @params2： durable 队列是否持久化
* @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
* @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
* @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
* */
channel.queueDeclare("ttl.queue2", true, false, false, null);
// 6： 准备发送消息的内容
String message = "你好，学相伴！！！";
Map<String, Object> headers = new HashMap<String, Object>();
headers.put("x", "1");
headers.put("y", "1");
AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder()
.deliveryMode(2) // 传送方式
.priority(1)
.contentEncoding("UTF-8") // 编码方式
.expiration("5000") // 过期时间
.headers(headers).build(); //自定义属性
// 7: 发送消息给中间件rabbitmq-server
// @params1: 交换机exchange
// @params2: 队列名称/routing
// @params3: 属性配置
// @params4: 发送消息的内容
for (int i = 0; i <10 ; i++) {
channel.basicPublish("", "ttl.queue2", basicProperties, message.getBytes());
System.out.println("消息发送成功!");
}
} catch (Exception ex) {
ex.printStackTrace();
System.out.println("发送消息出现异常...");
} finally {
// 7: 释放连接关闭通道
if (channel != null && channel.isOpen()) {
try {
channel.close();
} catch (Exception ex) {
ex.printStackTrace();
}
}
if (connection != null && connection.isOpen()) {
try {
connection.close();
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
}
}
3. 死信队列
DLX，全称为Dead-Letter-Exchange , 可以称之为死信交换机，也有人称之为死信邮箱。当消息在一个队列中变成死信(dead message)之后，它能被重新发送到另一个交换机中，这个交换机就是DLX ，绑定DLX的队列就称之为死信队列。
消息变成死信，可能是由于以下的原因：
- 消息被拒绝（没有被ack）
- 消息过期
- 队列达到最大长度
DLX也是一个正常的交换机，和一般的交换机没有区别，它能在任何的队列上被指定，实际上就是设置某一个队列的属性。当这个队列中存在死信时，Rabbitmq就会自动地将这个消息重新发布到设置的DLX上去，进而被路由到另一个队列，即死信队列。
要想使用死信队列，只需要在定义队列的时候设置队列参数 x-dead-letter-exchange 指定交换机即可。
3.1 创建死信队列
package com.zju.service.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @author guozonghao
* @Description
* @create 2024-10-04 09:37
*/
@Configuration
public class DeadRabbitMqConfiguration {
@Bean
public DirectExchange deadDirectExchange(){
return new DirectExchange("dead_direct_exchange",true,false);
}
@Bean
public Queue directDeadQueue(){
return new Queue("dead.direct.queue",true);
}
@Bean
public Binding directDeadBinding(){
return BindingBuilder.bind(directDeadQueue()).to(deadDirectExchange()).with("dead");
}
}
3.2 其他队列绑定死信队列
package com.zju.service.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @author guozonghao
* @Description
* @create 2024-10-04 09:37
*/
@Configuration
public class TTLRabbitMqConfiguration {
@Bean
public DirectExchange ttlDirectExchange(){
return new DirectExchange("ttl_direct_exchange",true,false);
}
@Bean
public Queue directTTLQueue(){
//设置过期时间
HashMap<String, Object> args = new HashMap<>();
args.put("x-message-ttl",5000);//设置过期时间
args.put("x-max-length",5);//设置队列的最大长度
args.put("x-dead-letter-exchange","dead_direct_exchange");//绑定自己创建的死信交换机
args.put("x-dead-letter-routing-key","dead");//设置死信队列的路由key
return new Queue("ttl.direct.queue",true,false,false,args);
}
@Bean
public Binding directTTLBinding(){
return BindingBuilder.bind(directTTLQueue()).to(ttlDirectExchange()).with("ttl");
}
}