1、死信的概念

2、死信来源

3、死信实战

3.1 代码架构

• 正常队列绑定正常交换机
• 正常队列绑定死信交换机
• 死信队列绑定死信

3.2 消息TTL过期变成死信

生产者向 normal_exchange发送消息，通过路由键zhangsan路由到 normal-queue中，消息设置TTL属性

/*** @author houChen* @date 2022/11/12 20:09* @Description: 死信队列实战： ttl* 生产者* 设置消息具有过期时间属性，当消息过期后会经过死信交换机路由到死信队列*/
public class Producer {private static final String NORMAL_EXECAGE = "normal_exchange";public static void main(String[] args) throws Exception {try (Channel channel = RabbitMqUtils.getChannel();) {//1、创建一个交换机channel.exchangeDeclare(NORMAL_EXECAGE, "direct");//设置消息的TTL时间AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().expiration("10000").build();for (int i = 0; i < 11; i++) {String message = "info" + i;channel.basicPublish(NORMAL_EXECAGE, "zhangsan", basicProperties, message.getBytes());System.out.println("生产者发送消息:" + message);}}}
}

消费者c1不启动，模拟消息在normal-queue中逗留超过10s，导致消息过期，经过dead_exchange路由到dead-queue

/*** @author houChen* @date 2022/11/12 20:33* @Description: 消费者c1代码*/
public class Consumer01 {private static final String NORMAL_EXCHANGE = "normal_exchange";private static final String DEAD_EXCHANGE = "dead_exchange";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//正常队列绑定死信交换机channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);Map params = new HashMap<>();//key是固定的params.put("x-dead-letter-exchange", DEAD_EXCHANGE);params.put("x-dead-letter-routing-key", "lisi");String normalQueueName = "normal-queue";channel.queueDeclare(normalQueueName, false, false, false, params);channel.queueBind(normalQueueName, NORMAL_EXCHANGE, "zhangsan");System.out.println("等待接收消息");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "utf-8");System.out.println("Consumer01接收到消息:" + message);};channel.basicConsume(normalQueueName, false, deliverCallback, consumerTag -> {});}
}

消费者c2消费死信队列中的消息

/*** @author houChen* @date 2022/11/12 21:11* @Description: Consumer02 会消费死信队列的消息*/
public class Consumer02 {private static final String DEAD_EXCHANGE = "dead_exchange";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);//声明死信队列,绑定String deadQueueName = "dead-queue";channel.queueDeclare(deadQueueName, false, false, false, null);channel.queueBind(deadQueueName, DEAD_EXCHANGE, "lisi");System.out.println("等待死信接收消息");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "utf-8");System.out.println("Consumer02接收死信队列消息:" + message);};channel.basicConsume(deadQueueName, true, deliverCallback, consumerTag -> {});}
}

启动生产者和消费者c2，发现经过10s后，消费者c2消费到生产者生产的消息，表明normal-queue中的消息过期后，确实经由dead-exchange 路由到dead-queue

3.3 队列达到最大长度

当 队列达到最大长度后，再往队列中投递消息时，消息会变成死信

1） 消息生产者代码去掉 TTL 属性：

/*** @author houChen* @date 2022/11/12 20:09* @Description: 死信队列实战： 队列达到最大长度导致消息进入死信队列* 生产者*  去掉消息的过期属性*/
public class Producer {private static final String NORMAL_EXECAGE = "normal_exchange";public static void main(String[] args) throws Exception {try (Channel channel = RabbitMqUtils.getChannel();) {//1、创建一个交换机channel.exchangeDeclare(NORMAL_EXECAGE, "direct");for (int i = 0; i < 10; i++) {String message = "info" + i;channel.basicPublish(NORMAL_EXECAGE, "zhangsan", null, message.getBytes());System.out.println("生产者发送消息:" + message);}}}
}

2） C1 消费者给 normal-queue 添加最大长度的属性 (启动之后关闭该消费者 模拟其接收不到消息)
【注意】 此时需要在RabbitMQ的控制台将 normal-queue删除，不然创建队列会报错

/*** @author houChen* @date 2022/11/12 20:33* @Description: 消费者c1代码*/
public class Consumer01 {private static final String NORMAL_EXCHANGE = "normal_exchange";private static final String DEAD_EXCHANGE = "dead_exchange";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);//正常队列绑定死信交换机Map params = new HashMap<>();//key是固定的//设置死信交换机params.put("x-dead-letter-exchange", DEAD_EXCHANGE);//设置死信路由键params.put("x-dead-letter-routing-key", "lisi");//设置正常队列的最大长度params.put("x-max-length", 6);//声明正常队列绑定正常交换机String normalQueueName = "normal-queue";channel.queueDeclare(normalQueueName, false, false, false, params);channel.queueBind(normalQueueName, NORMAL_EXCHANGE, "zhangsan");System.out.println("等待接收消息");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "utf-8");System.out.println("Consumer01接收到消息:" + message);};channel.basicConsume(normalQueueName, true, deliverCallback, consumerTag -> {});}
}

3）测试结果
消费者生产 10 条消息后，有 6 条消息进入dead-queue

3.4 消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false

1） 消息生产者代码同上生产者一致

2）C1 消费者代码 ： 对某个消息进行拒绝

/*** @author houChen* @date 2022/11/12 20:33* @Description: 消费者c1代码** 对 某个消息拒绝签收，并且不重新入队*/
public class Consumer01 {private static final String NORMAL_EXCHANGE = "normal_exchange";private static final String DEAD_EXCHANGE = "dead_exchange";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);//正常队列绑定死信交换机Map params = new HashMap<>();//key是固定的params.put("x-dead-letter-exchange", DEAD_EXCHANGE);params.put("x-dead-letter-routing-key", "lisi");//声明正常队列,绑定, 已经String normalQueueName = "normal-queue";channel.queueDeclare(normalQueueName, false, false, false, params);channel.queueBind(normalQueueName, NORMAL_EXCHANGE, "zhangsan");System.out.println("等待接收消息");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "utf-8");if(message.equals("info5")) {System.out.println("Consumer01接收到消息:" + message + "，并拒绝签收该消息");//param1: 消息的标记   param2： 被拒绝的消息是否重新入队channel.basicReject(delivery.getEnvelope().getDeliveryTag(),false);} else {System.out.println("Consumer01接收到消息:" + message);channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);}};channel.basicConsume(normalQueueName, false, deliverCallback, consumerTag -> {});}
}

3）C2 消费者代码不变
启动消费者 1 然后再启动消费者 2

4）结果
消费者c1,会拒绝消费消息info5，并且拒绝消息重新入队

被拒绝消息会路由到 dead-queue