死信

• 死信，当一个队列中的消息，满足以下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秒