RabbitMQ学习(二十一)：使用消息有效期TTL和死信路由DLX实现消息延迟重试消费的限制

我不是女神ヾ 2023-07-18 15:30 2阅读 0赞

# 说明 #

在之前的一篇博文[《springboot学习(十三)：RabbitMQ的使用 实现消息延迟消费》][springboot_RabbitMQ_]中，我简单介绍了使用Rabbitmq的消息有效期和死信路由的特性实现消息的延迟消费。在后续的使用中，希望通过这两种特性实现针对消息自定义过期时间实现延迟消费，但是发现Rabbitmq并不支持。本篇博文将对使用中出现的问题进行总结记录。

# 正文 #

在之前延迟消费一文的示例代码中，每个消息都设置了5s的有效期，在相同的过期时间时，消息的延迟消费可以正常实现。但是，在为每个消息设置不同的有效期时，会出现消息已过期却未被消费的情况。将之前的示例做以下修改：

## 生产者 ##

生产者在发送消息时，根据方法参数设置消息体中有效期livetime的值

public void sendEmail(String livetime) {
        JSONObject msg = new JSONObject();
        String message = "this is a email";
        msg.put("body", message);
        msg.put("livetime", livetime);
        String routingKey = "email";
        rabbitTemplate.convertAndSend(proExchange.getName(), routingKey, msg.toJSONString());
        System.out.println("send successfully");
    }

## 消费者 ##

消费者根据消息体中livetime参数值设置消息的有效期

@RabbitListener(queues = {"proQueue"})
    public void receiveEmail(String email) throws UnsupportedEncodingException {
        System.out.println("receive a email " + LocalDateTime.now().toString() + " " + email);
        JSONObject msg = JSON.parseObject(email);
        String livetime = msg.getString("livetime");
        String body = msg.getString("body");
        String routingKey = "email";
        MessageProperties messageProperties = getMessageProperties(livetime);
        if (body.indexOf("try") < 0) {
            JSONObject retry = new JSONObject();
            retry.put("body", body + " try again");
            retry.put("intervalTime", livetime);
            rabbitTemplate.convertAndSend(conExchange.getName(), routingKey, new Message((retry.toJSONString()).getBytes("utf-8"), messageProperties));
        }
    }

## 测试结果 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM2NTUzOTEz_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM2NTUzOTEz_size_16_color_FFFFFF_t_70 1]  
可以看到，当有效期短的消息先于有效期长的消息被发送时，消息能够正常的延迟重新消费，但是反之，有效期短的消息会被之前的消息阻塞而不能得到及时的死信路由。

## 原因 ##

这是因为在Rabbitmq中队列是消息的一个有序集合，消息按照FIFO的原则进行消费，在之前的一篇博文[《RabbitMQ学习(十二): 队列和消息的有效期 (Queue and Message TTL）》][RabbitMQ_ _ _Queue and Message TTL]中的注意事项中也提到，只有当过期消息到达队列头部时，过期消息才会被真正的丢弃或死信路由。

在官方文档中，对于消息的顺序性也有说明，一般情况下队列中的消息会按照FIFO的原则进行消费，但在优先级队列，分片队列或使用其他特性时，这种原则会被打破。

> Message Ordering  
> Queues in RabbitMQ are ordered collections of messages. Messages are enqueued and dequeued (consumed) in the FIFO manner, although priority queues, sharded queues and other features may affect this.

## 结论 ##

Rabbitmq中队列是一个FIFO类型的队列，消息在设置相同有效期时，能够按照目标实现延迟消息。但是在根据条件为消息设置不同的有效期时，因为消息只有在队列头部时，才会被死信路由，所以消息无法实现期望的及时准确的延迟消费。

那么还能使用RabbitMQ实现复杂的延迟消费的需求吗？  
答案是可以，可以通过RabbitMQ的rabbitmq\_delayed\_message\_exchange插件来设置每个消息的延迟时间。具体操作可以阅读这篇博文[《解决死信队列消息过期非异步问题，RabbitMQ 延时消息更优解——插件大法(Docker版)》][RabbitMQ _Docker]。

--------------------

参考资料：  
https://www.rabbitmq.com/queues.html\#message-ordering  
https://www.skypyb.com/2020/01/jishu/1206/

[springboot_RabbitMQ_]: https://blog.csdn.net/sinat_36553913/article/details/86433725
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM2NTUzOTEz_size_16_color_FFFFFF_t_70]: /images/20230528/6fd9ccf8588b41ae8f43a5b67351e94c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM2NTUzOTEz_size_16_color_FFFFFF_t_70 1]: /images/20230528/fb39c2fc81d043618e149e558804f2dd.png
[RabbitMQ_ _ _Queue and Message TTL]: https://blog.csdn.net/sinat_36553913/article/details/91351868
[RabbitMQ _Docker]: https://www.skypyb.com/2020/01/jishu/1323/