死信，顾名思义就是无法被消费的信息。一般来说，producer将消息投递到broker或者直接到queue里，consumer从queue取出消息进行消费，但某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信(Dead Letter)，所有的死信都会放到死信队列中。
应用场景：为了保证订单业务的消息数据不丢失，需要使用到RabbitMq的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。还有比如说：用户在商城下单成功并点击支付后在指定时间未支付时自动失效（可以用来做延迟消息）。
1、消息 TTL 过期。（TTL：存活时间，规定时间没被消费到）
2、队列达到最大长度(队列满了，无法再添加数据到 mq 中)。
3、消息被拒绝(basic.reject 或 basic.nack)并且 消息不重新放回对列（requeue=false）。
死信队列来保证消息至少被消费一次以及未被正确处理的消息不会被丢弃
2、代码架构图其中：zhangsan和lisi是routingKey
3、TTL过期生产者1
public class ProducerDeadQueue {
private static final String NORMAL_EXCHANGE_NAME = "normal_exchange_name";
public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
//延迟消息 设置ttl time to live 10s 如果这个消息超过10s没接收，就会移进死信队列中
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
for (int i=0; i<10;i++){
String message = "info " + i;
channel.basicPublish(NORMAL_EXCHANGE_NAME, "zhangsan", properties, message.getBytes("UTF-8"));
//我们可以在这里设置等待时间
}
channel.close();
connection.close();
}
}
普通路由与普通队列
public class ReceiveDeadQueue1 {
// 普通交换机的名称
public static final String NORMAL_EXCHANGE_NAME = "normal_exchange_name";
// 死信交换机
public static final String DEAD_EXCHANGE_NAME = "dead_exchange_name";
//普通队列
public static final String NORMAL_QUEUE_NAME = "normal_queue_name";
//死信队列
public static final String DEAD_QUEUE_NAME = "dead_queue_name";
public static void main(String[] argv) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
//声明死信和普通交换机，类型为direct
channel.exchangeDeclare(NORMAL_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
Map arguments = new HashMap<>();
//过期时间
//正常队列设设置死信交换机
//过期时间 10s=100000ms
// arguments.put("x-message-ttl",100000); //我们也可以在生产者中设置过期时间，会比较灵活
arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
//设置死信routingKey
arguments.put("x-dead-letter-routing-key", "lisi");
//声明普通队列 持久 共享 自动删除 arguments需要设置参数才能转发到死信队列
channel.queueDeclare(NORMAL_QUEUE_NAME, false, false, false, arguments);
//声明死信队列
channel.queueDeclare(DEAD_QUEUE_NAME, false, false, false, null);
//绑定普通的交换机与队列
channel.queueBind(NORMAL_QUEUE_NAME, NORMAL_EXCHANGE_NAME, "zhangsan");
//绑定死信的交换机与死信的队列
channel.queueBind(DEAD_QUEUE_NAME, DEAD_EXCHANGE_NAME, "lisi");
System.out.println("等待接收消息--------");
DeliverCallback deliverCallback = (consumerTag, message)->{
System.out.println("Consumer1 接受消息 : " + new String(message.getBody(), "UTF-8"));
System.out.println("接收队列："+ NORMAL_QUEUE_NAME +" , 绑定键： "+ message.getEnvelope().getRoutingKey());
};
channel.basicConsume(NORMAL_QUEUE_NAME, true, deliverCallback, consumerTag -> {});
}
}
消费者2监听死信队列
public class ReceiveDeadQueue2 {
//死信队列
public static final String DEAD_QUEUE_NAME = "dead_queue_name";
public static void main(String[] argv) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
System.out.println("等待接收消息--------");
DeliverCallback deliverCallback = (consumerTag, message)->{
System.out.println("Consumer2 接受消息 : " + new String(message.getBody(), "UTF-8"));
System.out.println("接收队列："+ DEAD_QUEUE_NAME +" , 绑定键： "+ message.getEnvelope().getRoutingKey());
};
channel.basicConsume(DEAD_QUEUE_NAME, true, deliverCallback, consumerTag -> {});
}
}
*** 作流程，我们先启动一下ReceiveDeadQueue1，让其产生对应的路由和队列
随后关闭
启动生产者1，不启动消费者1
等待10s后，会进入死信队列中
启动消费者2就可以接受到所有的消息
4、队列达到最大长度重点代码，在消费者1中加
arguments.put("x-max-length",6);//指定队列的最大长度，只能装6个消息，超过6个则成为死信
生产者
public class ProducerDeadQueue {
private static final String NORMAL_EXCHANGE_NAME = "normal_exchange_name";
public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
for (int i=0; i<10;i++){
String message = "info " + i;
channel.basicPublish(NORMAL_EXCHANGE_NAME, "zhangsan", null, message.getBytes("UTF-8"));
}
channel.close();
connection.close();
}
}
消费者1
public class ReceiveDeadQueue1 {
// 普通交换机的名称
public static final String NORMAL_EXCHANGE_NAME = "normal_exchange_name";
// 死信交换机
public static final String DEAD_EXCHANGE_NAME = "dead_exchange_name";
//普通队列
public static final String NORMAL_QUEUE_NAME = "normal_queue_name";
//死信队列
public static final String DEAD_QUEUE_NAME = "dead_queue_name";
public static void main(String[] argv) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
//声明死信和普通交换机，类型为direct
channel.exchangeDeclare(NORMAL_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
Map arguments = new HashMap<>();
//过期时间
//正常队列设设置死信交换机
//过期时间 10s=100000ms
// arguments.put("x-message-ttl",100000); //我们也可以在生产者中设置过期时间，会比较灵活
arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
//设置死信routingKey
arguments.put("x-dead-letter-routing-key", "lisi");
arguments.put("x-max-length",6);//指定队列的最大长度，只能装6个消息，超过6个则成为死信
//声明普通队列
channel.queueDeclare(NORMAL_QUEUE_NAME, false, false, false, arguments);
//声明死信队列
channel.queueDeclare(DEAD_QUEUE_NAME, false, false, false, null);
//绑定普通的交换机与队列
channel.queueBind(NORMAL_QUEUE_NAME, NORMAL_EXCHANGE_NAME, "zhangsan");
//绑定死信的交换机与死信的队列
channel.queueBind(DEAD_QUEUE_NAME, DEAD_EXCHANGE_NAME, "lisi");
System.out.println("等待接收消息--------");
DeliverCallback deliverCallback = (consumerTag, message)->{
System.out.println("Consumer1 接受消息 : " + new String(message.getBody(), "UTF-8"));
System.out.println("接收队列："+ NORMAL_QUEUE_NAME +" , 绑定键： "+ message.getEnvelope().getRoutingKey());
};
channel.basicConsume(NORMAL_QUEUE_NAME, true, deliverCallback, consumerTag -> {});
}
}
消费者2：死信消费者
public class ReceiveDeadQueue2 {
//死信队列
public static final String DEAD_QUEUE_NAME = "dead_queue_name";
public static void main(String[] argv) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
System.out.println("等待接收消息--------");
DeliverCallback deliverCallback = (consumerTag, message)->{
System.out.println("Consumer2 接受消息 : " + new String(message.getBody(), "UTF-8"));
System.out.println("接收队列："+ DEAD_QUEUE_NAME +" , 绑定键： "+ message.getEnvelope().getRoutingKey());
};
channel.basicConsume(DEAD_QUEUE_NAME, true, deliverCallback, consumerTag -> {});
}
}
实验启动，为了让消费者1假死，先将消费者1假死（关闭）
消费者2也假死（关闭）
生产者一启动
5、消费者进行拒绝
进行拒绝的 *** 作
if(str.equals("info 5")){
System.out.println("Consumer1 拒收消息： "+ str);
//C1进行拒绝
channel.basicReject(message.getEnvelope().getDeliveryTag(),false);//false 不放回原先队列，也可以塞回原先队列
}
生产者
public class ProducerDeadQueue {
private static final String NORMAL_EXCHANGE_NAME = "normal_exchange_name";
public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
for (int i=0; i<10;i++){
String message = "info " + i;
channel.basicPublish(NORMAL_EXCHANGE_NAME, "zhangsan", null, message.getBytes("UTF-8"));
//我们可以在这里设置等待时间
}
channel.close();
connection.close();
}
}
消费者1，且需要对对应的进行拒绝
public class ReceiveDeadQueue1 {
// 普通交换机的名称
public static final String NORMAL_EXCHANGE_NAME = "normal_exchange_name";
// 死信交换机
public static final String DEAD_EXCHANGE_NAME = "dead_exchange_name";
//普通队列
public static final String NORMAL_QUEUE_NAME = "normal_queue_name";
//死信队列
public static final String DEAD_QUEUE_NAME = "dead_queue_name";
public static void main(String[] argv) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
//声明死信和普通交换机，类型为direct
channel.exchangeDeclare(NORMAL_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
Map arguments = new HashMap<>();
//过期时间
//正常队列设设置死信交换机
//过期时间 10s=100000ms
// arguments.put("x-message-ttl",100000); //我们也可以在生产者中设置过期时间，会比较灵活
arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
//设置死信routingKey
arguments.put("x-dead-letter-routing-key", "lisi");
//声明普通队列
channel.queueDeclare(NORMAL_QUEUE_NAME, false, false, false, arguments);
//声明死信队列
channel.queueDeclare(DEAD_QUEUE_NAME, false, false, false, null);
//绑定普通的交换机与队列
channel.queueBind(NORMAL_QUEUE_NAME, NORMAL_EXCHANGE_NAME, "zhangsan");
//绑定死信的交换机与死信的队列
channel.queueBind(DEAD_QUEUE_NAME, DEAD_EXCHANGE_NAME, "lisi");
System.out.println("等待接收消息--------");
DeliverCallback deliverCallback = (consumerTag, message)->{
String str = new String(message.getBody(), "UTF-8");
if(str.equals("info 5")){
System.out.println("Consumer1 拒收消息： "+ str);
//C1进行拒绝
channel.basicReject(message.getEnvelope().getDeliveryTag(),false);//false 不放回原先队列，也可以塞回原先队列
}else {
System.out.println("Consumer1 接受消息 : " + str);
System.out.println("接收队列："+ NORMAL_QUEUE_NAME +" , 绑定键： "+ message.getEnvelope().getRoutingKey());
channel.basicAck(message.getEnvelope().getDeliveryTag(), false);//false 不批量
}
};
//开启手动应答
channel.basicConsume(NORMAL_QUEUE_NAME, false, deliverCallback, consumerTag -> {});
}
}
消费者2：死信队列
public class ReceiveDeadQueue2 {
//死信队列
public static final String DEAD_QUEUE_NAME = "dead_queue_name";
public static void main(String[] argv) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
System.out.println("等待接收消息--------");
DeliverCallback deliverCallback = (consumerTag, message)->{
System.out.println("Consumer2 接受消息 : " + new String(message.getBody(), "UTF-8"));
System.out.println("接收队列："+ DEAD_QUEUE_NAME +" , 绑定键： "+ message.getEnvelope().getRoutingKey());
};
channel.basicConsume(DEAD_QUEUE_NAME, true, deliverCallback, consumerTag -> {});
}
}
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）