在RabbiteMQ消息的拒绝 —— Reject和Nack中，我们是可以利用Reject和Nack去拒绝消息的，消息在被拒绝的时候是可以指定RabbitMQ是否需要重新发送给消费者。

如何我们消息被拒绝，且不重新投递的情况下，那么这条消息就无法被处理了么？在ActiveMQ死信队列中，我们介绍过消息失败被重发一定次数后会进入死信队列，在RabbitMQ中也有类似的处理——死信交换器。

死信交换器不是默认的设置，这里是被投递消息被拒绝后的一个可选行为，是在创建队列的时进行声明的，往往用在对问题消息的诊断上。

消息进入死信交换器一般是以下几种情况：

• 消息被拒绝，并且设置requeue参数为 false
• 消息过期
• 队列达到最大长度

死信交换器仍然只是一个普通的交换器，创建时并没有特别要求和操作。在创建队列的时候，声明该交换器将用作保存被拒绝的消息即可，相关的参数是x-dead-letter-exchange。

死信交换器的设置其实和RabbitMQ消息发布之备用交换器比较类似，不过备用交换器是消息生产者在发布消息时无法路由消息，消息将被路由到备用交换器，所以备用交换器配置在消息的生产者。而死信交换器则是接收过期或者被拒绝的消息，所以肯定是配置在消息的消费者。

首先我们先看看消息的发布者，这里和之前介绍的用法一样，就是发布了error、warn、info三条消息，如下：

public class Producer {//交换器名称public static final String EXCHANGE_NAME = "logs";public static void main(String[] args) throws Exception {//创建连接,连接到RabbitMQConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");Connection connection = connectionFactory.newConnection();//创建信道Channel channel = connection.createChannel();//创建交换器channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC, false, false, null);//定义的业务日志消息级别，即作为路由键使用String[] logLevels = {"error", "warn", "info"};for (String logLevel : logLevels) {String msg = "Hello RabbitMQ";//发布消息，需要参数：交换器、路由键，其中以日志消息级别为路由键channel.basicPublish(EXCHANGE_NAME, logLevel, null, msg.getBytes(Charset.forName("UTF-8")));}channel.close();connection.close();}
}

然后我们给生产者生成的三条消息，一共创建了2个消费者，一个是可以接收到全部的消息，并且全部拒绝并不重新投递，其中我们就需要配置其死信交换器了，如下：

public class AllConsumer {public static final String EXCHANGE_NAME = "logs";public static final String DLX_EXCHANGE_NAME = "dlx";public static void main(String[] args) throws Exception {//创建连接,连接到RabbitMQ,与发送端一样ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");Connection connection = connectionFactory.newConnection();//创建信道Channel channel = connection.createChannel();//创建交换器channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC, false, false, null);String queueName = "allLog";Map<String, Object> argMap = new HashMap<>();argMap.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);//创建一个队列channel.queueDeclare(queueName, false, false, false, argMap);//将队列和交换器通过路由键进行绑定channel.queueBind(queueName, EXCHANGE_NAME, "#");//声明了一个消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {String message = "Exchange: " + envelope.getExchange() + ", " +"RoutingKey: " + envelope.getRoutingKey() + ", " +"Content: " + new String(body, "UTF-8");System.out.println(message);channel.basicReject(envelope.getDeliveryTag(), false);}};//消费者正式开始在指定队列上消费消息channel.basicConsume(queueName, false, consumer);}
}

另外一个消费者，我们只接收其error消息，并且也是拒绝并不重新投递，不同的是我们将其送到死信交换器的时候，将其路由键进行重新定义了（死信交换器重新定义路由键，可以将其不同的消息，发送到同一个队列之类，可以进行统一处理失败的消息），代码如下：

public class ErrorConsumer {public static final String EXCHANGE_NAME = "logs";public static final String DLX_EXCHANGE_NAME = "dlx";public static void main(String[] args) throws Exception {//创建连接,连接到RabbitMQ,与发送端一样ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");Connection connection = connectionFactory.newConnection();//创建信道Channel channel = connection.createChannel();//创建交换器channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC, false, false, null);String queueName = "errorLog";Map<String, Object> argMap = new HashMap<>();argMap.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);//死信路由键，会替换消息原来的路由键argMap.put("x-dead-letter-routing-key", "log_error");//创建一个队列channel.queueDeclare(queueName, false, false, false, argMap);//将队列和交换器通过路由键进行绑定channel.queueBind(queueName, EXCHANGE_NAME, "error");//声明了一个消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {String message = "Exchange: " + envelope.getExchange() + ", " +"RoutingKey: " + envelope.getRoutingKey() + ", " +"Content: " + new String(body, "UTF-8");System.out.println(message);channel.basicReject(envelope.getDeliveryTag(), false);}};//消费者正式开始在指定队列上消费消息channel.basicConsume(queueName, false, consumer);}
}

上述两个消费者使用的是同一个死信交换器——dlx，两个消费者失败的消息都会进入同一个死信交换器中。当然我们也可以为每个消费者都定义一个单独的死信交换器，这样多个消费者失败的消息就会进入不同的死信交换器，我们在处理的时候就需要对多个死信交换器进行处理，这里就需要看其业务的要求的。

另外上述我们发现我们还没有创建死信交换器（当然我们也是可以在两个消费者中进行创建的，只要保证其参数相同），这里我们就是其死信交换器的消费者中进行创建的，这里同样我们为死信交换器也创建了两个消费者，一个可以消费其中的所有消息，如下：

public class AllDlxConsumer {public static final String DLX_EXCHANGE_NAME = "dlx";public static void main(String[] args) throws Exception {//创建连接,连接到RabbitMQ,与发送端一样ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");Connection connection = connectionFactory.newConnection();//创建信道Channel channel = connection.createChannel();//创建交换器channel.exchangeDeclare(DLX_EXCHANGE_NAME, BuiltinExchangeType.TOPIC, false, false, null);String queueName = channel.queueDeclare().getQueue();//将队列和交换器通过路由键进行绑定channel.queueBind(queueName, DLX_EXCHANGE_NAME, "#");//声明了一个消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {String message = "Exchange: " + envelope.getExchange() + ", " +"RoutingKey: " + envelope.getRoutingKey() + ", " +"Content: " + new String(body, "UTF-8");System.out.println(message);}};//消费者正式开始在指定队列上消费消息channel.basicConsume(queueName, true, consumer);}
}

另一个死信交换器消费者只用来处理那个被我们修改了路由键的error消息，上述中我们将其路由键有error修改为了log_error，其代码如下：

public class ErrorDlxConsumer {public static final String DLX_EXCHANGE_NAME = "dlx";public static void main(String[] args) throws Exception {//创建连接,连接到RabbitMQ,与发送端一样ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");Connection connection = connectionFactory.newConnection();//创建信道Channel channel = connection.createChannel();//创建交换器channel.exchangeDeclare(DLX_EXCHANGE_NAME, BuiltinExchangeType.TOPIC, false, false, null);String queueName = channel.queueDeclare().getQueue();//将队列和交换器通过路由键进行绑定channel.queueBind(queueName, DLX_EXCHANGE_NAME, "log_error");//声明了一个消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {String message = "Exchange: " + envelope.getExchange() + ", " +"RoutingKey: " + envelope.getRoutingKey() + ", " +"Content: " + new String(body, "UTF-8");System.out.println(message);}};//消费者正式开始在指定队列上消费消息channel.basicConsume(queueName, true, consumer);}
}

然后我运行上述所有代码，其结果如下，所有消费者受到error、warn、info所有消息，其实都是拒绝的

单独处理error消息的消费者，只收到其error消息，并拒绝其消息，将其路由键修改为log_error

上述四条消息拒绝都是正常的，所以都进入其死信交换器，我们先查看其消费所有死信交换器的消费者，如下

剩下的死信交换器消费者应该只是消费log_error的消息，如下：