死信队列

概念

推荐阅读1 推荐阅读2 推荐阅读3

死信队列全称 Dead-Letter-Exchange 简称 DLX 是 RabbitMQ 中交换器的一种类型，消息在一段时间之后没有被消费就会变成死信被重新 publish 到另一个 DLX 交换器队列中，因此称为死信队列。死信队列用于处理无法被正常消费的消息，即死信消息。

死信队列产生几种情况：

消息被拒绝(basic.reject / basic.nack)，并且requeue = false消息 TTL(Time To Live)(存活时间)过期队列达到最大长度

应用的场景

为了保证订单业务的消息数据不丢失，需要使用到RabbitMQ的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。

死信处理过程

DLX也是一个正常的Exchange，和一般的Exchange没有区别，它能在任何的队列上被指定，实际上就是设置某个队列的属性。当这个队列中有死信时，RabbitMQ就会自动的将这个消息重新发布到设置的Exchange上去，进而被路由到另一个队列。可以监听这个队列中的消息做相应的处理。

实战

消息TTL过期

消费者1

package com.uin.dead_Exchange;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.DeliverCallback;import com.uin.utils.RabbitMQUtils;import java.io.IOException;import java.util.HashMap;import java.util.concurrent.TimeoutException;public class Consumer1 { //普通的交换机 public static final String NORMAL_EXCHANGE = "normal_exchange2"; //死信的交换机 public static final String DEAD_EXCHANGE = "dead_exchange2"; //普通的队列 public static final String NORMAL_QUEUE = "normal_queue2"; //死信的队列 public static final String DEAD_QUEUE = "dead_queue2"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMQUtils.getChannel(); //声明交换机 channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); //普通队列要转发到死信交换机需要一些参数 arguments HashMap arguments = new HashMap<>(); //设置过期时间 //arguments.put("x-dead-ttl",10000); //设置正常队列为死信交换机 arguments.put("X-dead-letter-exchange", DEAD_EXCHANGE); //设置死信的routingkey arguments.put("X-dead-letter-routing-key", "lisi"); //声明队列 channel.queueDeclare(NORMAL_QUEUE, true, false, false,arguments ); channel.queueDeclare(DEAD_QUEUE, true, false, false, null); //普通的队列和普通的交换进行绑定 channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan"); //死信的交换机和死信的队列 进行绑定 channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi"); System.out.println("C1等待接受消息。。。"); //消费消息 DeliverCallback deliverCallback = (consumerTag, message) -> { String s = new String(message.getBody(), "UTF-8"); System.out.println("C1接收到的消息" + s); }; channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, consumerTag -> { System.out.println("消息取消消费的接口回调"); }); }}

生产者：

package com.uin.dead_Exchange;import com.rabbitmq.client.AMQP;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.uin.utils.RabbitMQUtils;import java.io.IOException;import java.nio.charset.StandardCharsets;import java.util.concurrent.TimeoutException;public class Producer { private static final String NORMAL_EXCHANGE = "normal_exchange2"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMQUtils.getChannel(); //声明交换机 channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); //设置消息的过期时间 TTL 参数 AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration( "1000").build(); //模拟发送消息 for (int i = 0; i < 10; i++) { String message = "info " + i; channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes(StandardCharsets.UTF_8)); } }}

队列的名称为2 。

消费者2

package com.uin.dead_Exchange;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.DeliverCallback;import com.uin.utils.RabbitMQUtils;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer2 { //死信的队列 public static final String DEAD_QUEUE = "dead_queue2"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMQUtils.getChannel(); System.out.println("C2等待接受消息。。。"); //消费消息 DeliverCallback deliverCallback = (consumerTag, message) -> { String s = new String(message.getBody(), "UTF-8"); System.out.println("C2接收到的消息" + s); }; channel.basicConsume(DEAD_QUEUE, true, deliverCallback, consumerTag -> { System.out.println("消息取消消费的接口回调"); }); }}

普通的队列将消息转发给死信队列接受的消息

队列达到最大长度

消费者代码的改动

生产者代码的改动：

运行c1，并断掉c1，运行生产者，运行死信的消费者c2

消息被拒 消息的生产者代码不改变 消息的消费者C1代码

package com.uin.dead_Exchange.max_length;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.DeliverCallback;import com.uin.utils.RabbitMQUtils;import java.io.IOException;import java.util.HashMap;import java.util.Map;import java.util.concurrent.TimeoutException;public class Consumer1 { //普通的交换机 public static final String NORMAL_EXCHANGE = "normal_exchange3"; //死信的交换机 public static final String DEAD_EXCHANGE = "dead_exchange3"; //普通的队列 public static final String NORMAL_QUEUE = "normal_queue3"; //死信的队列 public static final String DEAD_QUEUE = "dead_queue3"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMQUtils.getChannel(); //声明交换机 channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); //普通队列要转发到死信交换机需要一些参数 arguments Map arguments = new HashMap<>(); //设置过期时间 //arguments.put("x-dead-ttl",10000); //设置正常队列为死信交换机 arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE); //设置死信的routingkey arguments.put("x-dead-letter-routing-key", "lisi"); //设置队列放消息的最大长度// arguments.put("x-max-length", 6); //声明队列 channel.queueDeclare(NORMAL_QUEUE, true, false, false, arguments); channel.queueDeclare(DEAD_QUEUE, true, false, false, null); //普通的队列和普通的交换进行绑定 channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan"); //死信的交换机和死信的队列 进行绑定 channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi"); System.out.println("C1等待接受消息。。。"); //消费消息 DeliverCallback deliverCallback = (consumerTag, message) -> { String s = new String(message.getBody(), "UTF-8"); if (s.equals("info_message 5")) { System.out.println("Consumer01 接收到消息" + s + "并拒绝签收该消息"); //requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中 channel.basicReject(message.getEnvelope().getDeliveryTag(), false); } else { System.out.println("Consumer01 接收到消息" + s); channel.basicAck(message.getEnvelope().getDeliveryTag(), false); } }; channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, consumerTag -> { System.out.println("消息取消消费的接口回调"); }); }}

消息的消费者C2不变

package com.uin.dead_Exchange.reject_message;import com.rabbitmq.client.Channel;import com.rabbitmq.client.DeliverCallback;import com.uin.utils.RabbitMQUtils;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer2 { //死信的队列 public static final String DEAD_QUEUE = "dead_queue3"; public static void main(String[] args) throws IOException, TimeoutException { Channel channel = RabbitMQUtils.getChannel(); System.out.println("C2等待接受消息。。。"); //消费消息 DeliverCallback deliverCallback = (consumerTag, message) -> { String s = new String(message.getBody(), "UTF-8"); System.out.println("C2接收到的消息" + s); }; channel.basicConsume(DEAD_QUEUE, true, deliverCallback, consumerTag -> { System.out.println("消息取消消费的接口回调"); }); }}

测试结果：