死信
- 死信，当一个队列中的消息，满足以下3种某一种情况时，可以被成为死信（dead letter），即：要么处理不了的，要么不处理的，就叫死信
- 消费者，使用
basic.reject或basic.nack声明消费失败，并且消息的requeue参数设置为false - 消息是一个过期消息，超时无人消费
- 要投递的队列满了，放不下，无法投递
- 消费者，使用
Reject和Nack是什么？
- 消息确认，可以让RabbitMQ知道消费者已经接受并处理完消息。但是如果消息本身或者消息的处理过程出现问题怎么办？需要一种机制，通知RabbitMQ，这个消息，我无法处理，请让别的消费者处理。这里就有两种机制，Reject和Nack。
- Reject：Reject在拒绝消息时，可以使用
requeue标识，告诉RabbitMQ是否需要重新发送给别的消费者。不重新发送，一般这个消息就会被RabbitMQ丢弃。Reject一次只能拒绝一条消息。 - Nack：Nack则可以一次性拒绝多个消息。这是RabbitMQ对AMQP规范的一个扩展。
通过RejectRequeuConsumer可以看到当requeue参数设置为true时，消息发生了重新投递。
// RabbitMQ是否需要重新发送给别的消费者
channel.basicReject(envelope.getDeliveryTag(), false);
死信交换机
如果这个包含死信的队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机（Dead Letter Exchange，检查DLX）。
使用死信交换机，接收死信
- 定义死信交换机、死信队列、以及绑定2者
// 死信交换机
@Bean
public DirectExchange dlExchange() {
return new DirectExchange("dl.direct");
}
// 死信队列
@Bean
public Queue dlQueue() {
return new Queue("dl.queue");
}
// 绑定死信交换机和队列
@Bean
public Binding dlBinding() {
return BindingBuilder
// 队列
.bind(dlQueue())
// 交换机
.to(dlExchange())
// 配置routingKey
.with("dl");
}
- 指定队列的消息，如果被拒绝消费，或者队列满了，无法投递时，将消息投递到死信交换机
@Bean
public Queue simpleQueue() {
return QueueBuilder.durable("simple.queue").
deadLetterExchange("dl.direct").
deadLetterRoutingKey("dl").build();
}
- 修改消费者，消费异常时，reject拒绝消息，消息就会投递到死信交换机
@Component
@Slf4j
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(Message msg, Channel channel) throws Exception {
log.debug("--消费者接收到simple.queue的消息：【" + msg + "】");
// 模拟异常
try {
int i =1/0;
} catch (Exception e) {
channel.basicReject(msg.getMessageProperties().getDeliveryTag(),false);
}
log.debug("消息处理完成！");
}
}
TTL（Time To Live 生存时间值）
-
一个队列中的消息，如果超时未消费，就会变成死信，超时分为2种情况
- 消息所在的队列，设置了超时时间
- 消息本身设置了超时时间
定义死信交换机和死信队列，并配置TTL过期时间
@Configuration
public class TTLConfig {
// 声明死信交换机 dl.direct
@Bean
public DirectExchange ttlExchange(){
return new DirectExchange("ttl.direct", true, false);
}
// 声明存储死信的队列 dl.queue
@Bean
public Queue ttlQueue(){
// return new Queue("ttl.queue", true);
return QueueBuilder.durable("ttl.queue")
.deadLetterExchange("dl.ttl.exchange")
.deadLetterRoutingKey("dl")
.ttl(5000)
.build();
}
// 将死信队列与死信交换机绑定
@Bean
public Binding ttlBinding(){
return BindingBuilder
// 队列
.bind(ttlQueue())
// 交换机
.to(ttlExchange())
// routingKey
.with("ttl");
}
}
- 定义消费者，并声明死信交换机和死信队列、routingKey等
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "dl.ttl.queue", durable = "true"),
exchange = @Exchange(name = "dl.ttl.exchange"),
key = "dl"
))
public void listenDlQueue(String msg){
log.info("接收到 dl.ttl.queue的延迟消息：{}", msg);
}
- 生产者发送消息到MQ
@Test
public void testTTLMessageQueue() {
try {
String routingKey = "ttl";
String message = "hello, spring amqp!";
rabbitTemplate.convertAndSend("ttl.direct", routingKey, message);
} catch (Exception e) {
e.printStackTrace();
}
}
总结
- 消息超时的2种方式
- 给队列设置
ttl属性，进入队列后超过ttl时间的消息变为死信 - 给消息设置
ttl属性，队列接收到消息超过ttl时间后变为死信
- 给队列设置
- 如何实现发送一个消息20秒后消费者才收到消息？（3个步骤）
- 给消息的目标队列，指定死信交换机
- 将消费者监听的队列，绑定到死信交换机
- 发送消息时，给消息设置超时时间为20秒
