RabbitMQ实践——超时消息的处理方法

news2024/11/24 0:24:10

大纲

准备工作
整个队列的消息都有相同的时效性
- 抛弃超时消息
- - 新建带x-message-ttl的队列
  - 新建绑定关系
  - 实验
- 超时消息路由到死信队列
- - 新建带死信和ttl的队列
  - 新建绑定关系
  - 实验
消息指定自己的超时时间
- 新建带死信的队列
- 绑定
- 实验
消息自带TTL和队列TTL的关系
- 消息TTL < 队列指定TTL
- 消息TTL > 队列指定TTL
- 总结

在项目中，我们往往会遇到消息具有时效性：在一定时间内的消息需要处理，其他消息则不用处理。RabbitMQ提供了两种功能来满足这种需求：

整个队列的消息都有相同的时效性。
消息可以指定自己的超时时间。

对于超时的消息，我们可以选择抛弃，或者让其进入死信。本文我们将实验这些场景。

准备工作

我们先新建一个交换器direct.ttl。后续不同场景我们再设定其不同路由。
在这里插入图片描述
然后再创建一个死信队列queue.dead.letter

最后新建交换器direct.ttl和队列queue.dead.letter的路由绑定关系

后续我们将根据不同场景新建不同的队列和绑定关系。

整个队列的消息都有相同的时效性

抛弃超时消息

新建带x-message-ttl的队列

如果队列设置了x-message-ttl，则其全部消息的最大ttl就是其值。它的单位是毫秒。下图中，队列消息如果30秒没有处理，就会从队列中移除。
在这里插入图片描述

新建绑定关系

我们让交换器中Routing key是to.queue.all.message.ttl的消息路由到上面创建的队列中。
在这里插入图片描述

实验

向该交换器发送一条消息
在这里插入图片描述
消息出现在队列queue.all.message.ttl中。

半分钟后，该条消息从队列中消失。

超时消息路由到死信队列

不同于上例，只是设置了x-message-ttl，还要设置死信相关参数。本例我们将使用“重写Routing key”的方案，节省一次交换器的创建。

新建带死信和ttl的队列

新建名字是queue.all.msg.ttl.dead.letter的队列。这个队列中的所有消息的最大超时时间（x-message-ttl）是30秒；超时消息会被修改Routing key为to.queue.dead.letter后，使用direct.ttl交换器来路由。
在这里插入图片描述

新建绑定关系

将这个队列和之前新建的交换器direct.ttl关联。
在这里插入图片描述

实验

在这里插入图片描述

可以看到消息在过了30秒后，被从原来的队列queue.all.msg.ttl.dead.letter中移除，然后路由到死信队列queue.dead.letter中。

消息指定自己的超时时间

本例我们只实验相对复杂的场景，即将超时消息路由到死信。

新建带死信的队列

这个队列我们没有设置ttl相关数据，只是设置了死信相关配置：死信消息交由交换器direct.ttl处理，其Routing key被修改成to.queue.dead.letter。
在这里插入图片描述

绑定

在这里插入图片描述

实验

这次我们不能使用后台来发送消息，而是通过Java代码来发送。

public void sendWithTTL(String exchangeName, String routingKey, String message, int ttl) {
    String msgId = UUID.randomUUID().toString();
    Message msg = MessageBuilder.withBody(message.getBytes())
            .setContentType("text/plain")
            .setCorrelationId(msgId)
            .setMessageId(msgId)
            .setExpiration(String.valueOf(ttl))
            .build();

    CorrelationData correlationData = new CorrelationData(msgId);
    rabbitTemplateWithoutMandatory.convertAndSend(exchangeName, routingKey, msg, correlationData);
}