【RabbitMQ的那点事】死信列队（Dead Letter Q

2023-02-04 本文已影响0人伊丽莎白2015

【官方文档】

Dead Letter Exchanges

【参考】

https://www.cloudamqp.com/blog/when-and-how-to-use-the-rabbitmq-dead-letter-exchange.html

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);
    }
}

打印日志：

可以看到消息在队列里先是呆了10秒钟，没有被消费，过期后才移入了死信队列。

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);
    }
}

运行测试用例，一共发了5条消息，其中2条在queue中：

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);
    }

可以看到第1条消息由于一直被requeue，导致消费端消费被阻塞了，第2条消息是无法被消费的。

image.png

这时候也可以消费端通过判断是否是重新设递的message：message.getMessageProperties().getRedelivered()来决定下一次的requeue情况。

不过总体来说这跟死信队列无关，对于死信队列来说，它关心的是requeue=false状态下的否定应答的消息。

【RabbitMQ的那点事】死信列队（Dead Letter Q

1. dead-lettered定义

2. 处理的方式

3. 如何配置Dead Letter Exchange

4. 使用Spring Boot来实现Dead Letter Exchange功能

4.1 case-1: 消息被consumer端接收到后，consumer端使用了negatively acknowledge（否定应答），即：`basic.reject`或`basic.nack`，并且参数requeue=false。

【在RabbitMQ Console中查看】

4.2 case-2: 设置了TTL（time-to-live）的消息过期了。

4.3 case-3: queue超出了设置的size。

5. 问题：为什么在#4.1中消费端给出否定应答时，需要参数requeue=false？

猜你喜欢

热点阅读

【RabbitMQ的那点事】死信列队（Dead Letter Q

1. dead-lettered定义

2. 处理的方式

3. 如何配置Dead Letter Exchange

4. 使用Spring Boot来实现Dead Letter Exchange功能

4.1 case-1: 消息被consumer端接收到后，consumer端使用了negatively acknowledge（否定应答），即：basic.reject或basic.nack，并且参数requeue=false。

【在RabbitMQ Console中查看】

4.2 case-2: 设置了TTL（time-to-live）的消息过期了。

4.3 case-3: queue超出了设置的size。

5. 问题：为什么在#4.1中消费端给出否定应答时，需要参数requeue=false？

猜你喜欢

热点阅读

4.1 case-1: 消息被consumer端接收到后，consumer端使用了negatively acknowledge（否定应答），即：`basic.reject`或`basic.nack`，并且参数requeue=false。