【官方文档】
【参考】
1. dead-lettered定义
什么情况下的消息可被当作“dead-lettered（死信）”：
- 消息被consumer端接收到后，consumer端使用了negatively acknowledge（否定应答），即：
basic.reject或basic.nack，并且参数requeue=false时。 - 设置了TTL（time-to-live）的消息过期了。
- 当queue超出了设置的size（即队列满了）。
2. 处理的方式
- 默认情况下，RabbitMQ broker会把上述情况的message给丢掉。Consumer端并不会正常的消费掉这些消息。
-
也可配置Dead Letter Exchange来收集上述的消息。有了这个Dead Letter Exchange，消息将被转到Dead Letter Queue（死信队列）中，下一步怎么处理死信队列中的消息，取决于应用的逻辑。
image.png
3. 如何配置Dead Letter Exchange
Dead Letter Exchange在定义上和别的普通的exchange没有什么不同。比如我们定义一个Direct Exchange叫dlx_exchange，用来接收Dead Letter Message：
channel.exchangeDeclare("dlx_exchange", "direct");
那么如何才能让Dead Letter Message从普通的queue中跳转到该死信exchange上呢？比如我们普通的Queue叫user_queue。在定义user_queue的时候，需要传入参数：
-
x-dead-letter-exchange：我们定义的dead letter exchange的名字。
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-dead-letter-exchange", "dlx_exchange");
channel.queueDeclare("user_queue", false, false, false, args);
还可以设置routingKey，比如等于dlx_key，这里的routingKey的意思是上述的dlx_exchange发送到Dead Letter Queue时的routingKey。如果不填，默认等于上一个routingKey。
args.put("x-dead-letter-routing-key", "dlx_key");
整个flow可以用下图来表示，本来正常的流程只有user_exchange, user_queue，routingKey=user.key。
我们尝试在user_queue中加入死信队列，这时候需要设置x-dead-letter-exchange，如果不设置x-dead-letter-routing-key，那么dlx_exchange和dl_queuebinding时的routingKey则和正常流程中的routingKey一致，即user.key，如果设置了，那么则使用设置的routingKey：
Dead Letter Exchange
4. 使用Spring Boot来实现Dead Letter Exchange功能
4.1 case-1: 消息被consumer端接收到后，consumer端使用了negatively acknowledge（否定应答），即：basic.reject或basic.nack，并且参数requeue=false。
-
userQueue()方法，定义queue=user_queue的时候，额外定义了它的Dead Letter Exchange。 -
receiveUserMessage()方法，user_queue的消费端，正常情况下给于positive ack，即正确的应答：channel.basicAck(tag, false)，但如果消息包含dead letter字符串，则给于negative ack，即第#1章的basic.reject，方法hannel.basicReject(tag, false)第2个参数为requeue，这里要设置为false。
@Slf4j
@Configuration
public class DeadLetterConfig {
// define normal:
private static final String normalExchangeName = "user_exchange";
private static final String normalQueueName = "user_queue";
private static final String normalRoutingKey = "user.key";
// define dead letter:
private static final String dlExchangeName = "dl_exchange";
private static final String dlQueueName = "dl_queue";
private static final String dlRoutingKey = "dlx";
@Bean
public DirectExchange userDirectExchange() {
return new DirectExchange(normalExchangeName);
}
@Bean
public Queue userQueue() {
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", dlExchangeName);
arguments.put("x-dead-letter-routing-key", dlRoutingKey);
return new Queue(normalQueueName, true, false, false, arguments);
}
@Bean
public Binding userDirectBinding(Queue userQueue, DirectExchange userDirectExchange) {
return BindingBuilder.bind(userQueue).to(userDirectExchange).with(normalRoutingKey);
}
@RabbitListener(queues = normalQueueName)
public void receiveUserMessage(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
String messageBody = new String(message.getBody());
log.info("Get message: {}", messageBody);
if (messageBody.contains("dead letter")) {
channel.basicReject(tag, false);
} else {
channel.basicAck(tag, false);
}
}
// ------------------------------------------------------------
@Bean
public DirectExchange deadLetterDirectExchange() {
return new DirectExchange(dlExchangeName);
}
@Bean
public Queue deadLetterQueue() {
return new Queue(dlQueueName, true);
}
@Bean
public Binding deadLetterDirectBinding(Queue deadLetterQueue, DirectExchange deadLetterDirectExchange) {
return BindingBuilder.bind(deadLetterQueue).to(deadLetterDirectExchange).with(dlRoutingKey);
}
@RabbitListener(queues = dlQueueName)
public void receiveDeadLetterMessage(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
log.info("Received Dead Letter Message: {}", new String(message.getBody()));
channel.basicAck(tag, false);
}
}
另外还要在application.yaml中设置下应答模式为manual：
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: manual
【测试】
@SpringBootTest
public class DeadLetterTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void send() throws InterruptedException {
rabbitTemplate.convertAndSend("user_exchange", "user.key", "hello.");
rabbitTemplate.convertAndSend("user_exchange", "user.key", "hello, i am dead letter messsage.");
Thread.sleep(5000);
}
}
运行后，可以看到第1条消息被正常处理，第2条消息因为应答模式为basic.reject并且requeue=false，则被转到了Dead Letter Exchange中：
image.png
【在RabbitMQ Console中查看】
在RabbitMQ Console中查看我们定义的user_queue，可以看到多了DLX和DLK，分别代表x-dead-letter-exchange和x-dead-letter-routing-key。
image.png
而我们定义的Dead Letter Exchange=dlx_exchange与正常的exchange无异，我们也可以换成别的type，如fanout或topic exchange：
image.png
4.2 case-2: 设置了TTL（time-to-live）的消息过期了。
代码和#4.1差不多，我改了下exchange和queue的名字。
// define normal:
private static final String normalExchangeName = "menu_exchange";
private static final String normalQueueName = "menu_queue";
private static final String normalRoutingKey = "menu.key";
// define dead letter:
private static final String dlExchangeName = "dl_menu_exchange";
private static final String dlQueueName = "dl_menu_queue";
private static final String dlRoutingKey = "menu.dlx";
在测试的时候可以把消费端关掉，让消息在队列里过期。
测试代码，需要设置消息的过期时间：
@Slf4j
@SpringBootTest
public class DeadLetterTTLTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void send() throws InterruptedException {
log.info("start to send message.");
for (int i = 0; i < 5; i ++) {
MessageProperties properties = new MessageProperties();
properties.setExpiration("10000");
Message message = new Message(("message: " + i).getBytes(StandardCharsets.UTF_8), properties);
rabbitTemplate.send("menu_exchange", "menu.key", message);
}
Thread.sleep(500000);
}
}
打印日志：
image.png
还有一种方式是不需要在发送消息的时候设置过期时间，而是将ttl设置在queue中也是可以的：
@Bean
public Queue menuQueue() {
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", dlExchangeName);
arguments.put("x-dead-letter-routing-key", dlRoutingKey);
arguments.put("x-message-ttl", 10000);
return new Queue(normalQueueName, true, false, false, arguments);
}
4.3 case-3: queue超出了设置的size。
代码和#4.1差不多，我改了下exchange和queue的名字。
// define normal:
private static final String normalExchangeName = "role_exchange";
private static final String normalQueueName = "role_queue";
private static final String normalRoutingKey = "role.key";
// define dead letter:
private static final String dlExchangeName = "dl_role_exchange";
private static final String dlQueueName = "dl_role_queue";
另外，在normalQueue的创建的时候需要设置容量大小：
@Bean
public Queue roleQueue() {
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", dlExchangeName);
arguments.put("x-max-length", 2);
return new Queue(normalQueueName, true, false, false, arguments);
}
在测试的时候可以把消费端关掉，让消息在队列里先不要被消费，这样才会使queue容量满掉。
测试代码：
@SpringBootTest
public class DeadLetterQueueMaxLengthTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void send() throws InterruptedException {
for (int i =0; i < 5; i ++) {
rabbitTemplate.convertAndSend("role_exchange", "role.key", "hello, i am dead letter messsage: " + i);
}
Thread.sleep(1000000);
}
}
image.png
另外3条因为队列容量满了会被移到死信队列中：
image.png
5. 问题：为什么在#4.1中消费端给出否定应答时，需要参数requeue=false？
原因也比较简单，因为requeue=true的时候，这条消息会被重新放回queue。详见：https://www.rabbitmq.com/nack.html
比如我们在#4.1的消费端requeue置为true时：
@RabbitListener(queues = normalQueueName)
public void receiveUserMessage(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
String messageBody = new String(message.getBody());
log.info("Get message: {}", messageBody);
if (messageBody.contains("dead letter")) {
channel.basicReject(tag, true);
} else {
channel.basicAck(tag, false);
}
}
测试代码：
@Test
public void send() throws InterruptedException {
rabbitTemplate.convertAndSend("user_exchange", "user.key", "hello dead letter messsage.");
rabbitTemplate.convertAndSend("user_exchange", "user.key", "hi~~~~~~~.");
Thread.sleep(100);
}
image.png
这时候也可以消费端通过判断是否是重新设递的message：message.getMessageProperties().getRedelivered()来决定下一次的requeue情况。
不过总体来说这跟死信队列无关，对于死信队列来说，它关心的是requeue=false状态下的否定应答的消息。
网友评论