实现延时队列，可以通过RabbitMQ的死信队列（Dead-letter queue）特性，“死信队列”是当消息过期，或者队列达到最大长度时，未消费的消息会被加入到死信队列。然后，我们可以对死信队列中的消息进行消费，完成类似“延时”的效果。
下面的示例代码演示了如何在Spring Boot中使用RabbitMQ设置一个订单，然后在15分钟后自动取消。
1. 添加 RabbitMQ 依赖:
在你的pom.xml中加入Spring Boot对RabbitMQ的Starter：
xml
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2. 配置队列、交换器、绑定和容器
我们创建一个配置类，定义一个正常的队列和一个死信队列，以及相应的交换机和队列的绑定：
java
@Configuration
public class RabbitMQConfig {
/* 正常队列配置 */
@Bean
public Queue orderQueue() {
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dead_exchange"); // 队列消息过期后发送的交换器
args.put("x-dead-letter-routing-key", "dead"); // 队列消息过期后发送的路由键
return new Queue("order_queue", true, false, false, args);
}
@Bean
public DirectExchange orderExchange() {
return new DirectExchange("order_exchange");
}
@Bean
public Binding orderBinding() {
return BindingBuilder.bind(orderQueue()).to(orderExchange()).with("order");
}
/* 死信队列配置 */
@Bean
public Queue deadQueue() {
return new Queue("dead_queue");
}
@Bean
public DirectExchange deadExchange() {
return new DirectExchange("dead_exchange");
}
@Bean
public Binding deadBinding() {
return BindingBuilder.bind(deadQueue()).to(deadExchange()).with("dead");
}
}
3. 发送延时消息
创建订单时，我们发送一个延时消息到队列：
java
@Autowired
private RabbitTemplate rabbitTemplate;
public void createOrder(String orderId) {
//订单创建业务...
rabbitTemplate.convertAndSend(
"order_exchange",
"order",
orderId,
message -> {
message.getMessageProperties().setExpiration(String.valueOf(15 * 60 * 1000)); // 15分钟
return message;
}
);
}
4. 消费死信队列中的消息
然后，我们需要消费死信队列中的消息，进行订单取消的操作：
java
@RabbitListener(queues = "dead_queue")
public void processDeadLetter(String orderId) {
// 订单取消业务...
}
- 在RabbitMQ中设置消息的过期时间
RabbitMQ允许你在两个级别上设置消息的过期时间：队列级别和消息级别。一旦消息过期，它将从队列中删除。
- 队列级别：你可以在声明队列时通过"x-message-ttl"参数设置过期时间（以毫秒为单位）。这会影响队列中的所有消息。
java
Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 60000); // 设置60s过期时间
Channel channel = ...;
channel.queueDeclare("my_queue", false, false, false, args);
- 消息级别：你也可以在发布消息时单独为每条消息设置过期时间。如果队列级别的TTL和消息级别的TTL都被设置了，那么较小的那个值会被应用。
java
byte[] messageBodyBytes = "Hello, world!".getBytes();
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().expiration("60000").build(); // 设置60s过期时间
Channel channel = ...;
channel.basicPublish("my_exchange", "my_routing_key", props, messageBodyBytes);
5、RabbitMQ中的死信队列如何工作？
“死信队列”用于接收不可路由的消息，或者由于一些原因不能正确处理的消息。这样能防止原始队列堵塞，也可以进一步处理这些消息。
当以下情况发生时，消息会被投递到死信队列：
- 消息被拒绝 (basic.reject / basic.nack)，并且requeue=false。
- 在队列中排队的时间过长（超过设置的TTL）。
- 队列长度溢出(超过设置的最大长度)。
在声明队列时，可以通过"x-dead-letter-exchange"和"x-dead-letter-routing-key"来设置死信交换器和路由键。
6、如何确保消费者在消费消息时不会发生重复消费的情况？
确保消费者在消费消息时不发生重复消费，一般可以通过以下方式实现：
- 消息确认机制：RabbitMQ提供了消息确认机制ACK，消费者处理完消息后，返回一个ACK信号，告诉RabbitMQ可以将该消息从队列中删除。如果在处理消息的过程中消费者出现异常（如断开连接），RabbitMQ将不会删除队列中的消息，并且会再次尝试发送该消息给消费者。
- 幂等操作：在消费者端，确保消息处理逻辑是幂等的，也就是说，无论消息被处理一次还是多次，结果都是相同的。例如，对于一个扣款操作，无论这个操作执行多少次，账户的余额都应该只被扣除一次。
- 分布式锁：在处理消息之前，消费者首先尝试获取一个分布式锁。只有获取到锁的消费者才能处理消息。处理完毕后释放锁。避免了同一消息被多个消费者处理的情况。
7、设置RabbitMQ中消息的优先级
在RabbitMQ中，对消息的优先级的支持是通过队列来实现的。在声明队列的时候，可以通过x-max-priority参数来指定队列支持的最大优先级。然后在发布消息的时候，可以通过basicProperties的priority字段来指定消息的优先级。如下：
java
// 声明队列时设置最大优先级
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-max-priority", 10);
channel.queueDeclare("my_queue", false, false, false, args);
// 发布消息时设置优先级
AMQP.BasicProperties props = new AMQP.BasicProperties().builder().priority(5).build();
channel.basicPublish("", "my_queue", props, "Hello world".getBytes());
注意，优先级较高的消息将会优先被消费，但是并不保证完全按照优先级顺序消费。
8、RabbitMQ中如何处理消费者异常断开连接的情况？
当RabbitMQ检测到消费者（如一个TCP连接）异常断开，例如因为消费者主机崩溃或因为网络问题，它将关闭该消费者的连接，并将消费者未确认的任何消息重新放入队列。如果你希望一个消息在消费者断开连接时不被再次放入队列，你可以设置该消费者的autoAck参数为true（也就是无需显示确认）。但一般情况下，我们推荐消费者在正确处理消息后发送一个确认应答（basicAck）。
