我们先新建一个 死信队列交换机 exchange-dead
“死信”是RabbitMQ中的一种消息机制，当你在消费消息时，如果队列里的消息出现以下情况：
以上三种情况的消息，将成为“死信”，“死信”消息会被RabbitMQ进行特殊处理，如果配置了死信队列信息，那么该消息将会被丢进死信队列中，如果没有配置，则该消息将会被丢弃。
先定义交换机类型，然后我们分别演示一下几种情况的死信队列
声明交换机信息
package dead;
public enum ExchangeTypeEnum {
DIRECT("exchange-direct-name", "direct"),
FANOUT("exchange-fanout-name", "fanout"),
TOPIC("exchange-topic-name", "topic"),
HEADER("exchange-header-name", "header"),
UNKNOWN("unknown-exchange-name", "direct");
/**
* 交换机名字
*/
private String name;
/**
* 交换机类型
*/
private String type;
ExchangeTypeEnum(String name, String type) {
this.name = name;
this.type = type;
}
public String getName() {
return name;
}
public String getType() {
return type;
}
public static ExchangeTypeEnum getEnum(String type) {
ExchangeTypeEnum[] exchangeArrays = ExchangeTypeEnum.values();
for (ExchangeTypeEnum exchange : exchangeArrays) {
if (exchange.getName().equals(type)) {
return exchange;
}
}
return ExchangeTypeEnum.UNKNOWN;
}
}
我们生产一个消息，然后消费者Reject处理，然后看下消息处理方式
首先我们先理解一下 消息被拒绝丢弃
channel.basicNack(deliveryTag, multiple, requeue);
下面我们看下用法
channel.basicReject(deliveryTag, requeue);
相比channel.basicNack，除了没有multiple批量确认机制之外，其他语义完全一样。
新建一个死信队列常量类
package dead;
public class DeadConst {
/**
* 死信交换机
*/
public static final String DEAD_EXCHANGE = "exchange_dead";
/**
* 新建一个队列，用于接受通过 死信交换机路由过来的 死信消息
*/
public static final String DEAD_MSG_QUEUE = "dead_msg_queue";
/**
* 设置 rk=# 表示任意RK的消息过来，都可以路由到 dead_msg_quque这个队列
*/
public static final String RK_DERAD_MSG_QUEUE = "#";
}
执行 deadInit()方法
package dead;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import conn.MqConnectUtil;
import subscrib3.ExchangeTypeEnum;
import java.time.LocalDate;
import java.time.LocalTime;
import java.util.HashMap;
import java.util.Map;
import static dead.DeadConst.*;
public class DeadReject {
/**
* 队列名字
*/
public final static String REJECT_QUEUE_NAME = "reject_queue_test";
/**
* routingkey
*/
public final static String RK_REJECT_QUEUE_NAME = "rk.reject_queue_test";
/**
* 声明死信队列信息
*
* @throws Exception
*/
public static void deadInit() throws Exception {
// 获取到连接以及mq通道
Connection connection = MqConnectUtil.getConnectionDefault();
// 从连接中创建通道
Channel channel = connection.createChannel();
/*声明 直连交换机 交换机 String exchange,
* 参数明细
* 1、交换机名称
* 2、交换机类型，topic
*/
channel.exchangeDeclare(DEAD_EXCHANGE, ExchangeTypeEnum.TOPIC.getType());
channel.queueDeclare(DEAD_MSG_QUEUE, true, false, false, null);
/*交换机和队列绑定String queue, String exchange, String routingKey
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key rk.dead_msg_queue
*/
channel.queueBind(DEAD_MSG_QUEUE, DEAD_EXCHANGE, RK_DERAD_MSG_QUEUE);
//关闭通道和连接
channel.close();
connection.close();
}
/**
* 生产 Direct直连 交换机的MQ消息
*/
public static void produce() throws Exception {
// 获取到连接以及mq通道
Connection connection = MqConnectUtil.getConnectionDefault();
// 从连接中创建通道
Channel channel = connection.createChannel();
/*声明 直连交换机 交换机 String exchange,
* 参数明细
* 1、交换机名称
* 2、交换机类型，direct
*/
channel.exchangeDeclare(ExchangeTypeEnum.DIRECT.getName(), ExchangeTypeEnum.DIRECT.getType());
/* 声明（创建）队列 queueDeclare( String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
* queue - 队列名
* durable - 是否是持久化队列， 队列的声明默认是存放到内存中的，如果rabbitmq重启会丢失
* exclusie - 是否排外的，仅限于当前队列使用
* autoDelete - 是否自动删除队列，当最后一个消费者断开连接之后队列是否自动被删除，可以通过界面 查看某个队列的消费者数量，当consumers = 0时队列就会自动删除
* arguments - 队列携带的参数 比如 ttl-生命周期，x-dead-letter 死信队列等等
*/
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
// arguments.put("x-dead-letter-routing-key", RK_REJECT_QUEUE_NAME);
arguments.put("x-dead-letter-routing-key", "xxx");
channel.queueDeclare(REJECT_QUEUE_NAME, true, false, false, arguments);
/*交换机和队列绑定String queue, String exchange, String routingKey
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key rk.subscribe_queue_direct
*/
channel.queueBind(REJECT_QUEUE_NAME, ExchangeTypeEnum.DIRECT.getName(), RK_REJECT_QUEUE_NAME);
/* 发送消息 String exchange, String routingKey, BasicProperties props, byte[] body
* exchange - 交换机 ，"" 空时候指定的是 获取的virtualHost 虚拟服务器的 默认的exchang，每个virtualHost都有一个AMQP default type:direct 直接转发
* queuename - 队列信息
* props - 参数信息
* message 消息体 byte[]类型
*/
// 消息内容
String message = "i=1" + " Hello World! Time:" + LocalDate.now() + " " + LocalTime.now();
channel.basicPublish(ExchangeTypeEnum.DIRECT.getName(), RK_REJECT_QUEUE_NAME, null, message.getBytes());
System.out.println(" **** Producer Sent Message: [" + message + "]");
//关闭通道和连接
channel.close();
connection.close();
}
public static void consumer() throws Exception {
Connection connection = null;
Channel channel = null;
try {
connection = MqConnectUtil.getConnectionDefault();
channel = connection.createChannel();
/*声明交换机 String exchange
* 参数明细
* 1、交换机名称
* 2、交换机类型，fanout、topic、direct、headers
*/
channel.exchangeDeclare(ExchangeTypeEnum.DIRECT.getName(), ExchangeTypeEnum.DIRECT.getType());
//交换机和队列绑定String queue, String exchange, String routingKey
/**
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key
*/
channel.queueBind(REJECT_QUEUE_NAME, ExchangeTypeEnum.DIRECT.getName(), RK_REJECT_QUEUE_NAME);
System.out.println(" **** Consumer->1 Waiting for messages. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
/* 消息确认机制
* autoAck true:表示自动确认，只要消息从队列中获取，无论消费者获取到消息后是否成功消费，都会认为消息已经成功消费
* autoAck false:表示手动确认，消费者获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，如果消费者一直没有反馈，那么该消息将一直处于不可用状态
* 并且服务器会认为该消费者已经挂掉，不会再给其发送消息，直到该消费者反馈
* !!!!!! 注意这里是 false，手动确认
*/
channel.basicConsume(REJECT_QUEUE_NAME, false, consumer);
int count = 0;
while (count < 10) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" count:" + count + " **** Consumer->2 Received '" + message + "'");
doSomeThing(message);
//返回tag的当前标签，不再重新路由，直接丢弃
channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
count++;
}
} catch (Exception e) {
e.printStackTrace();
} finally {
channel.close();
connection.close();
}
}
/**
* 模拟处理复杂逻辑：休眠100ms
*
* @param message
* @throws Exception
*/
public static void doSomeThing(String message) throws Exception {
//遍历Count ，sleep ， 接收一条消息后休眠 100 毫秒，模仿复杂逻辑
Thread.sleep(100);
}
public static void main(String[] args) throws Exception {
//初始化死信队列
deadInit();
// //生产消息
// produce();
// //消费者
// consumer();
}
}
我们看下 死信交换机exchange_dead和 接受死亡消息的队列
public static void main(String[] args) throws Exception {
//初始化死信队列
// deadInit();
// //生产消息
produce();
// //消费者
// consumer();
}
执行 produce() 生产1条消息
执行consumer()
public static void main(String[] args) throws Exception {
//初始化死信队列
// deadInit();
// //生产消息
// produce();
// //消费者
consumer();
}
!!! 注意 consumer的消费者 通过
channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
来拒绝消息，第二个参数表示单个tag的丢弃、第三个参数表示不再重新路由到队列
看一下原始队列
我们看一下死亡信息存放的队列，发现有1条消息，正式我们刚才的consumer拒绝的那条消息
至此~ 通过消费者Reject，到死信交换机，在通过绑定的RoutingKey：# 路由的过程已经完毕
清空、删除之前的队列信息，继续
下面我们试一下，多次路由的情况，构造一种场景
队列A发消息->A的消费者拒绝->通过Reject的第三个参数 requeue=true 重新路由->再到当前队列A
取出消息，看下消息的参数到底有何改变
删除之前的死信交换机和之前的队列信息
package dead;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import conn.MqConnectUtil;
import subscrib3.ExchangeTypeEnum;
import java.time.LocalDate;
import java.time.LocalTime;
public class QueueMessageRouteTest {
/**
* 队列名字
*/
public final static String REJECT_QUEUE_NAME = "reject_queue_test";
/**
* routingkey
*/
public final static String RK_REJECT_QUEUE_NAME = "rk.reject_queue_test";
/**
* 生产 Direct直连 交换机的MQ消息
*/
public static void produce() throws Exception {
// 获取到连接以及mq通道
Connection connection = MqConnectUtil.getConnectionDefault();
// 从连接中创建通道
Channel channel = connection.createChannel();
/*声明 直连交换机 交换机 String exchange,
* 参数明细
* 1、交换机名称
* 2、交换机类型，direct
*/
channel.exchangeDeclare(subscrib3.ExchangeTypeEnum.DIRECT.getName(), subscrib3.ExchangeTypeEnum.DIRECT.getType());
/* 声明（创建）队列 queueDeclare( String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
* queue - 队列名
* durable - 是否是持久化队列， 队列的声明默认是存放到内存中的，如果rabbitmq重启会丢失
* exclusie - 是否排外的，仅限于当前队列使用
* autoDelete - 是否自动删除队列，当最后一个消费者断开连接之后队列是否自动被删除，可以通过界面 查看某个队列的消费者数量，当consumers = 0时队列就会自动删除
* arguments - 队列携带的参数 比如 ttl-生命周期，x-dead-letter 死信队列等等
*/
channel.queueDeclare(REJECT_QUEUE_NAME, true, false, false, null);
/*交换机和队列绑定String queue, String exchange, String routingKey
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key rk.subscribe_queue_direct
*/
channel.queueBind(REJECT_QUEUE_NAME, subscrib3.ExchangeTypeEnum.DIRECT.getName(), RK_REJECT_QUEUE_NAME);
/* 发送消息 String exchange, String routingKey, BasicProperties props, byte[] body
* exchange - 交换机 ，"" 空时候指定的是 获取的virtualHost 虚拟服务器的 默认的exchang，每个virtualHost都有一个AMQP default type:direct 直接转发
* queuename - 队列信息
* props - 参数信息
* message 消息体 byte[]类型
*/
// 消息内容
String message = "i=1" + " Hello World! Time:" + LocalDate.now() + " " + LocalTime.now();
channel.basicPublish(subscrib3.ExchangeTypeEnum.DIRECT.getName(), RK_REJECT_QUEUE_NAME, null, message.getBytes());
System.out.println(" **** Producer Sent Message: [" + message + "]");
//关闭通道和连接
channel.close();
connection.close();
}
public static void consumer() throws Exception {
Connection connection = null;
Channel channel = null;
try {
connection = MqConnectUtil.getConnectionDefault();
channel = connection.createChannel();
/*声明交换机 String exchange
* 参数明细
* 1、交换机名称
* 2、交换机类型，fanout、topic、direct、headers
*/
channel.exchangeDeclare(subscrib3.ExchangeTypeEnum.DIRECT.getName(), subscrib3.ExchangeTypeEnum.DIRECT.getType());
//交换机和队列绑定String queue, String exchange, String routingKey
/**
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key
*/
channel.queueBind(REJECT_QUEUE_NAME, ExchangeTypeEnum.DIRECT.getName(), RK_REJECT_QUEUE_NAME);
System.out.println(" **** Consumer->1 Waiting for messages. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
/* 消息确认机制
* autoAck true:表示自动确认，只要消息从队列中获取，无论消费者获取到消息后是否成功消费，都会认为消息已经成功消费
* autoAck false:表示手动确认，消费者获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，如果消费者一直没有反馈，那么该消息将一直处于不可用状态
* 并且服务器会认为该消费者已经挂掉，不会再给其发送消息，直到该消费者反馈
* !!!!!! 注意这里是 false，手动确认
*/
channel.basicConsume(REJECT_QUEUE_NAME, false, consumer);
int count = 0;
while (count < 10) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" count:" + count + " **** Consumer->2 Received '" + message + "'");
doSomeThing(message);
//返回tag的当前标签，不再重新路由，直接丢弃
channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
count++;
}
} catch (Exception e) {
e.printStackTrace();
} finally {
channel.close();
connection.close();
}
}
/**
* 模拟处理复杂逻辑：休眠100ms
*
* @param message
* @throws Exception
*/
public static void doSomeThing(String message) throws Exception {
//遍历Count ，sleep ， 接收一条消息后休眠 100 毫秒，模仿复杂逻辑
Thread.sleep(100);
}
public static void main(String[] args) throws Exception {
// //生产消息
// produce();
// //消费者
// consumer();
}
}
执行produce生产1条消息，然后同reject 来控制重新路由，启动下消费者
!!! 注意消费者我循环了10次，如果会重新路由，肯定会有多次消费消费的
而且队列中始终有这一条消息存在
所以为了避免出现我们无限消费的情况，对消息的幂等性进行处理，我们要对消息设置唯一ID，消费过的消息再次过来时候，拒绝消费
清空、删除之前的队列信息，继续
消息设置生命周期，超过生命周期的消息，会通过死信交换机、再次路由到死信队列
初始化死信队列 deadInit()，生产消息执行produce()，我们设置TTL=30s
package dead;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import conn.MqConnectUtil;
import subscrib3.ExchangeTypeEnum;
import java.time.LocalDate;
import java.time.LocalTime;
import java.util.HashMap;
import java.util.Map;
import static dead.DeadConst.*;
public class DeadTtlDelete {
/**
* 队列名字
*/
public final static String TTL_QUEUE_NAME = "ttl_queue_test";
/**
* routingkey
*/
public final static String RK_TTL_QUEUE_NAME = "rk.ttl_queue_test";
/**
* 声明死信队列信息
*
* @throws Exception
*/
public static void deadInit() throws Exception {
// 获取到连接以及mq通道
Connection connection = MqConnectUtil.getConnectionDefault();
// 从连接中创建通道
Channel channel = connection.createChannel();
/*声明 直连交换机 交换机 String exchange,
* 参数明细
* 1、交换机名称
* 2、交换机类型，topic
*/
channel.exchangeDeclare(DEAD_EXCHANGE, ExchangeTypeEnum.TOPIC.getType());
channel.queueDeclare(DEAD_MSG_QUEUE, true, false, false, null);
/*交换机和队列绑定String queue, String exchange, String routingKey
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key rk.dead_msg_queue
*/
channel.queueBind(DEAD_MSG_QUEUE, DEAD_EXCHANGE, RK_DERAD_MSG_QUEUE);
//关闭通道和连接
channel.close();
connection.close();
}
/**
* 生产 Direct直连 交换机的MQ消息
*/
public static void produce() throws Exception {
// 获取到连接以及mq通道
Connection connection = MqConnectUtil.getConnectionDefault();
// 从连接中创建通道
Channel channel = connection.createChannel();
/*声明 直连交换机 交换机 String exchange,
* 参数明细
* 1、交换机名称
* 2、交换机类型，direct
*/
channel.exchangeDeclare(ExchangeTypeEnum.DIRECT.getName(), ExchangeTypeEnum.DIRECT.getType());
/* 声明（创建）队列 queueDeclare( String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
* queue - 队列名
* durable - 是否是持久化队列， 队列的声明默认是存放到内存中的，如果rabbitmq重启会丢失
* exclusie - 是否排外的，仅限于当前队列使用
* autoDelete - 是否自动删除队列，当最后一个消费者断开连接之后队列是否自动被删除，可以通过界面 查看某个队列的消费者数量，当consumers = 0时队列就会自动删除
* arguments - 队列携带的参数 比如 ttl-生命周期，x-dead-letter 死信队列等等
*/
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
arguments.put("x-dead-letter-routing-key", "xxx");
arguments.put("x-message-ttl", 30000);
channel.queueDeclare(TTL_QUEUE_NAME, true, false, false, arguments);
/*交换机和队列绑定String queue, String exchange, String routingKey
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key rk.subscribe_queue_direct
*/
channel.queueBind(TTL_QUEUE_NAME, ExchangeTypeEnum.DIRECT.getName(), RK_TTL_QUEUE_NAME);
/* 发送消息 String exchange, String routingKey, BasicProperties props, byte[] body
* exchange - 交换机 ，"" 空时候指定的是 获取的virtualHost 虚拟服务器的 默认的exchang，每个virtualHost都有一个AMQP default type:direct 直接转发
* queuename - 队列信息
* props - 参数信息
* message 消息体 byte[]类型
*/
// 消息内容
String message = "i=1" + " Hello World! Time:" + LocalDate.now() + " " + LocalTime.now();
channel.basicPublish(ExchangeTypeEnum.DIRECT.getName(), RK_TTL_QUEUE_NAME, null, message.getBytes());
System.out.println(" **** Producer Sent Message: [" + message + "]");
//关闭通道和连接
channel.close();
connection.close();
}
public static void main(String[] args) throws Exception {
//初始化死信队列
deadInit();
// //生产消息
produce();
}
}
执行后我们看下队列信息
30s过后，原始队列中的消息死亡消失，之前死信队列没消息，现在出现了一条消息
至此~ 设置TTL超时死亡的消息，路由到死信队列完毕
清空、删除之前的队列信息，继续
先初始化死信队列deadInit()，然后执行produce()，设置队列长度为1，但是我们生产3条消息，看下多出的2条消息如何处理
package dead;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import conn.MqConnectUtil;
import subscrib3.ExchangeTypeEnum;
import java.time.LocalDate;
import java.time.LocalTime;
import java.util.HashMap;
import java.util.Map;
import static dead.DeadConst.*;
public class DeadMaxLengthReject {
/**
* 队列名字
*/
public final static String MAX_LENGTH_QUEUE_NAME = "max_length_queue_test";
/**
* routingkey
*/
public final static String RK_MAX_LENGTH_QUEUE_NAME = "rk.max_length_queue_test";
/**
* 声明死信队列信息
*
* @throws Exception
*/
public static void deadInit() throws Exception {
// 获取到连接以及mq通道
Connection connection = MqConnectUtil.getConnectionDefault();
// 从连接中创建通道
Channel channel = connection.createChannel();
/*声明 直连交换机 交换机 String exchange,
* 参数明细
* 1、交换机名称
* 2、交换机类型，topic
*/
channel.exchangeDeclare(DEAD_EXCHANGE, ExchangeTypeEnum.TOPIC.getType());
channel.queueDeclare(DEAD_MSG_QUEUE, true, false, false, null);
/*交换机和队列绑定String queue, String exchange, String routingKey
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key rk.dead_msg_queue
*/
channel.queueBind(DEAD_MSG_QUEUE, DEAD_EXCHANGE, RK_DERAD_MSG_QUEUE);
//关闭通道和连接
channel.close();
connection.close();
}
/**
* 生产 Direct直连 交换机的MQ消息
*/
public static void produce(Integer i) throws Exception {
// 获取到连接以及mq通道
Connection connection = MqConnectUtil.getConnectionDefault();
// 从连接中创建通道
Channel channel = connection.createChannel();
/*声明 直连交换机 交换机 String exchange,
* 参数明细
* 1、交换机名称
* 2、交换机类型，direct
*/
channel.exchangeDeclare(ExchangeTypeEnum.DIRECT.getName(), ExchangeTypeEnum.DIRECT.getType());
/* 声明（创建）队列 queueDeclare( String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
* queue - 队列名
* durable - 是否是持久化队列， 队列的声明默认是存放到内存中的，如果rabbitmq重启会丢失
* exclusie - 是否排外的，仅限于当前队列使用
* autoDelete - 是否自动删除队列，当最后一个消费者断开连接之后队列是否自动被删除，可以通过界面 查看某个队列的消费者数量，当consumers = 0时队列就会自动删除
* arguments - 队列携带的参数 比如 ttl-生命周期，x-dead-letter 死信队列等等
*/
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
arguments.put("x-dead-letter-routing-key", "xxx");
arguments.put("x-max-length", 1);
channel.queueDeclare(MAX_LENGTH_QUEUE_NAME, true, false, false, arguments);
/*交换机和队列绑定String queue, String exchange, String routingKey
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key rk.subscribe_queue_direct
*/
channel.queueBind(MAX_LENGTH_QUEUE_NAME, ExchangeTypeEnum.DIRECT.getName(), RK_MAX_LENGTH_QUEUE_NAME);
/* 发送消息 String exchange, String routingKey, BasicProperties props, byte[] body
* exchange - 交换机 ，"" 空时候指定的是 获取的virtualHost 虚拟服务器的 默认的exchang，每个virtualHost都有一个AMQP default type:direct 直接转发
* queuename - 队列信息
* props - 参数信息
* message 消息体 byte[]类型
*/
// 消息内容
String message = "i" + i + " Hello World! Time:" + LocalDate.now() + " " + LocalTime.now();
channel.basicPublish(ExchangeTypeEnum.DIRECT.getName(), RK_MAX_LENGTH_QUEUE_NAME, null, message.getBytes());
System.out.println(" **** Producer Sent Message: [" + message + "]");
//关闭通道和连接
channel.close();
connection.close();
}
public static void main(String[] args) throws Exception {
//初始化死信队列
deadInit();
for (int i = 0; i < 3; i++) {
// 生产消息
produce(i);
}
}
}
看下生产消息后，队列中的结果
可以看到i=0和i=1超出了队列长度，导致被死信交换机路由到死信队列中
至此~ 超出队列长度导致死信队列路由的完毕
下一篇 我们讲一下 RabbitMQ系列（十一）RabbitMQ进阶-Queue队列详解-延时队列