15、RabbitMQ 实战 - 死信队列的简介与死信队列和死信消费者的实现
1、 死信的概念；
死信，顾名思义就是无法被消费的消息，一般来说producer（生产者）将消息投递到broker或直接放到queue（队列）中，consumer（消费者）从queue（队列）取出消息进行消费，但某些时候由于特定的原因导致queue（队列）中的消息无法被消费，若这些消息没有后续的处理，则这些消息就变成了死信，有死信自然就有了死信队列
2、 死信的应用场景；
为保证订单业务的消息数据不丢失，需要使用RabbitMQ的死信队列机制，当消息发生异常时，将消息投入死信队列中
3、 死信的来源；
(1)消息TTL（存活时间）过期
(2)队列达到最大长度（队列满了，无法再添加数据到mq中）
(3)消息被拒绝（basic.reject或basic.nack）并且requeue=false
4、 死信队列的实现；
(1)我们将根据这张死信队列的代码架构图来实现死信队列
(2)新建一个名为dead的包，用于装实现死信队列的代码
效果图：
(3)新建一个名为Consumer01的类用于编写消费者的代码
代码如下：
注：RabbitMqUtils工具类的实现在我的另一篇文章里，有需要的同学可以查看参考
RabbitMQ系列（6）--RabbitMQ模式之工作队列(Work queues)的简介及实现_Ken_1115的博客-CSDN博客
package com.ken.dead;
import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
import java.util.HashMap;
import java.util.Map;
public class Consumer01 {
//普通交换机的名称
public static final String NORMAL_EXCHANGE = "normal_exchange";
//死信交换机的名称
public static final String DEAD_EXCHANGE = "dead_exchange";
//普通队列的名称
public static final String NORMAL_QUEUE = "normal_queue";
//死信队列的名称
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
//声明普通死信交换机，类型为direct
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
//声明死信交换机，类型为direct
channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
//用于在消息成为死信后，把消息转发到死信交换机dead_exchange里
Map<String, Object> arguments = new HashMap<>();
//正常队列设置死信交换机
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
//设置死信routingkey
arguments.put("x-dead-letter-routing-key","dead");
//设置正常队列长度的限制
//arguments.put("x-max-length",6);
/**
* 声明普通队列
* 第一个参数：队列名称
* 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
* 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
* 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
* 第五个参数：其他参数
*/
channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
/**
* 声明死信队列
* 第一个参数：队列名称
* 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
* 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
* 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
* 第五个参数：其他参数
*/
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
//普通队列与普通交换机通过routingkey进行捆绑
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"normal");
//死信队列与死信交换机通过routingkey进行捆绑
channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"dead");
/**
* 声明消费者接收消息后的回调方法(由于回调方法DeliverCallback是函数式接口，所以需要给DeliverCallback赋值一个函数，为了方便我们这里使用Lambda表达式进行赋值)
* 为什么要这样写呢，是因为basicConsume方法里的参数deliverCallback的类型DeliverCallback用 @FunctionalInterface注解规定DeliverCallback是一个函数式接口，所以要往deliverCallback参数传的值要是一个函数
*
* 以下是DeliverCallback接口的源代码
* @FunctionalInterface
* public interface DeliverCallback {
* void handle (String consumerTag, Delivery message) throws IOException;
* }
*/
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("Consumer01接收的消息是：" + new String(message.getBody(),"UTF-8"));
};
channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,consumerTag -> {});
}
}
(4)复制Consumer01类并粘贴重命名为Consumer02
代码如下：
package com.ken.dead;
import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.util.HashMap;
import java.util.Map;
public class Consumer02 {
//普通队列的名称
public static final String NORMAL_QUEUE = "normal_queue";
//死信队列的名称
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
/**
* 声明消费者接收消息后的回调方法(由于回调方法DeliverCallback是函数式接口，所以需要给DeliverCallback赋值一个函数，为了方便我们这里使用Lambda表达式进行赋值)
* 为什么要这样写呢，是因为basicConsume方法里的参数deliverCallback的类型DeliverCallback用 @FunctionalInterface注解规定DeliverCallback是一个函数式接口，所以要往deliverCallback参数传的值要是一个函数
*
* 以下是DeliverCallback接口的源代码
* @FunctionalInterface
* public interface DeliverCallback {
* void handle (String consumerTag, Delivery message) throws IOException;
* }
*/
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("Consumer02接收的消息是：" + new String(message.getBody(),"UTF-8"));
};
channel.basicConsume(DEAD_QUEUE,true,deliverCallback,consumerTag -> {});
}
}
(5)新建一个名为Producer的类用于编写生产者的代码
代码如下：
package com.ken.dead;
import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
/**
* 生产者
*/
public class Producer {
//普通交换机的名称
public static final String NORMAL_EXCHANGE = "normal_exchange";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
//死信消息，设置TTL时间（存活时间），单位是ms 10000ms = 10s
AMQP.BasicProperties properties= new AMQP.BasicProperties().builder().expiration("10000").build();
for (int i = 1; i < 11; i++) {
String message = "info" + i;
/**
* 用信道对消息进行发布（消息持久化）
* 第一个参数：发送到哪个交换机
* 第二个参数：路由的Key值是哪个，本次是队列名
* 第三个参数：其他参数信息
* 第四个参数：发送消息的消息体
*/
channel.basicPublish(NORMAL_EXCHANGE,"normal",properties,message.getBytes());
}
}
}
(6)先运行Consumer01，生成队列和交换机
(7)然后停止Consumer01，模拟消费者宕机
(8)运行Producer
(9)观察normal_queue队列和dead_queue队列消息数量的变化，一开始normal_queue队列里有10条消息，过了10s后消息都到了dead_queue队列里，证明消费者消费消息失败，消息从normal_queue队列移到了dead_queue队列里，由此可见当消息TTL过期后，死信队列成功运行
(10)启动Consumer02，可以看到Consumer02消费了dead_queue队列里的消息
(10)删除normal_queue队列
(11) 修改Produces的代码，把设置TTL时间的代码注释掉，basicPublish方法的第3个参数设置为null
代码如下：
package com.ken.dead;
import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
/**
* 生产者
*/
public class Producer {
//普通交换机的名称
public static final String NORMAL_EXCHANGE = "normal_exchange";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
//死信消息，设置TTL时间（存活时间），单位是ms 10000ms = 10s
//AMQP.BasicProperties properties= new AMQP.BasicProperties().builder().expiration("10000").build();
for (int i = 1; i < 11; i++) {
String message = "info" + i;
/**
* 用信道对消息进行发布（消息持久化）
* 第一个参数：发送到哪个交换机
* 第二个参数：路由的Key值是哪个，本次是队列名
* 第三个参数：其他参数信息
* 第四个参数：发送消息的消息体
*/
channel.basicPublish(NORMAL_EXCHANGE,"normal",null,message.getBytes());
}
}
}
(12)修改Consumer01的代码，把之前注释的用于限制队列长度的代码放出来
代码如下：
package com.ken.dead;
import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
import java.util.HashMap;
import java.util.Map;
public class Consumer01 {
//普通交换机的名称
public static final String NORMAL_EXCHANGE = "normal_exchange";
//死信交换机的名称
public static final String DEAD_EXCHANGE = "dead_exchange";
//普通队列的名称
public static final String NORMAL_QUEUE = "normal_queue";
//死信队列的名称
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
//声明普通死信交换机，类型为direct
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
//声明死信交换机，类型为direct
channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
//用于在消息成为死信后，把消息转发到死信交换机dead_exchange里
Map<String, Object> arguments = new HashMap<>();
//正常队列设置死信交换机
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
//设置死信routingkey
arguments.put("x-dead-letter-routing-key","dead");
//设置正常队列长度的限制
arguments.put("x-max-length",6);
/**
* 声明普通队列
* 第一个参数：队列名称
* 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
* 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
* 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
* 第五个参数：其他参数
*/
channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
/**
* 声明死信队列
* 第一个参数：队列名称
* 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
* 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
* 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
* 第五个参数：其他参数
*/
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
//普通队列与普通交换机通过routingkey进行捆绑
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"normal");
//死信队列与死信交换机通过routingkey进行捆绑
channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"dead");
/**
* 声明消费者接收消息后的回调方法(由于回调方法DeliverCallback是函数式接口，所以需要给DeliverCallback赋值一个函数，为了方便我们这里使用Lambda表达式进行赋值)
* 为什么要这样写呢，是因为basicConsume方法里的参数deliverCallback的类型DeliverCallback用 @FunctionalInterface注解规定DeliverCallback是一个函数式接口，所以要往deliverCallback参数传的值要是一个函数
*
* 以下是DeliverCallback接口的源代码
* @FunctionalInterface
* public interface DeliverCallback {
* void handle (String consumerTag, Delivery message) throws IOException;
* }
*/
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("Consumer01接收的消息是：" + new String(message.getBody(),"UTF-8"));
};
channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,consumerTag -> {});
}
}
(13)停掉所有程序，然后重新运行Consumer01，生成队列和交换机 
(14)然后停止Consumer01，模拟消费者宕机
(15)重新运行Producer
(16)观察normal_queue队列和dead_queue队列消息数量的变化，可以看到normal_queue队列只能堆积最多6条消息，而剩余的4条消息都移到了dead_queue队列里，由此可见当队列达到最大长度6条后，死信队列成功运行
(17)删除normal_queue队列和dead_queue队列
(18)修改Consumer01的代码，注释掉限制队列长度的代码，修改消息应答的部分代码
代码如下：
package com.ken.dead;
import com.ken.utils.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
import java.util.HashMap;
import java.util.Map;
public class Consumer01 {
//普通交换机的名称
public static final String NORMAL_EXCHANGE = "normal_exchange";
//死信交换机的名称
public static final String DEAD_EXCHANGE = "dead_exchange";
//普通队列的名称
public static final String NORMAL_QUEUE = "normal_queue";
//死信队列的名称
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
//声明普通死信交换机，类型为direct
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
//声明死信交换机，类型为direct
channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
//用于在消息成为死信后，把消息转发到死信交换机dead_exchange里
Map<String, Object> arguments = new HashMap<>();
//正常队列设置死信交换机
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
//设置死信routingkey
arguments.put("x-dead-letter-routing-key","dead");
//设置正常队列长度的限制
//arguments.put("x-max-length",6);
/**
* 声明普通队列
* 第一个参数：队列名称
* 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
* 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
* 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
* 第五个参数：其他参数
*/
channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
/**
* 声明死信队列
* 第一个参数：队列名称
* 第二个参数：服务器重启后队列是否还存在，即队列是否持久化,true为是，false为否，默认false，即消息存储在内存中而不是硬盘中
* 第三个参数：该队列是否只供一个消费者进行消费，是否进行消息共享，true为只允许一个消费者进行消费，false为允许多个消费者对队列进行消费，默认false
* 第四个参数：是否自动删除，最后一个消费者断开连接后该队列是否自动删除，true自动删除，false不自动删除
* 第五个参数：其他参数
*/
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
//普通队列与普通交换机通过routingkey进行捆绑
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"normal");
//死信队列与死信交换机通过routingkey进行捆绑
channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"dead");
/**
* 声明消费者接收消息后的回调方法(由于回调方法DeliverCallback是函数式接口，所以需要给DeliverCallback赋值一个函数，为了方便我们这里使用Lambda表达式进行赋值)
* 为什么要这样写呢，是因为basicConsume方法里的参数deliverCallback的类型DeliverCallback用 @FunctionalInterface注解规定DeliverCallback是一个函数式接口，所以要往deliverCallback参数传的值要是一个函数
*
* 以下是DeliverCallback接口的源代码
* @FunctionalInterface
* public interface DeliverCallback {
* void handle (String consumerTag, Delivery message) throws IOException;
* }
*/
DeliverCallback deliverCallback = (consumerTag, message) -> {
String mes = new String(message.getBody(),"UTF-8");
//消息被拒绝，basicReject第二个参数是false表示被拒绝后不放回队列
if("info5".equals(mes)) {
System.out.println("Consumer01接收的消息是：" + mes + "，此消息被拒绝");
channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
}else {
System.out.println("Consumer01接收的消息是：" + mes);
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
}
};
//开启手动应答
channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,consumerTag -> {});
}
}
(19)停掉所有程序，然后重新运行Consumer01，生成队列和交换机 
(20)重新运行Producer
(21)查看Consumer01控制台的输出，观察dead_queue队列消息数量的变化，因为没有开启Consumer02消费dead_queue队列，可以看到dead_queue队列堆积了1条消息，查看这条消息，可以看出就是我们拒绝掉的info5，这证明消费者拒绝消费消息info5后，消息info5从normal_queue队列移到了dead_queue队列里，由此可见当消息被拒绝消费后，死信队列成功运行
(22)启动Consumer02，可以看到Consumer02消费了dead_queue队列里的消息
