资讯

历史

科技

环境与自然

成长

游戏

财经

文学与艺术

美食

健康

家居

文化

情感

汽车

三农

军事

旅行

运动

教育

生活

星座命理

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

创作时间:

作者:

@小白创作中心

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

引用

CSDN

https://blog.csdn.net/breaksoftware/article/details/139727392

在项目中，我们往往会遇到消息具有时效性：在一定时间内的消息需要处理，其他消息则不用处理。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);
}

发送消息时我们只需要给消息增加setExpiration调用即可。它的单位是毫秒，需要转换为字符串。
可以看到20秒后，消息从queue.with.dead.letter队列进入queue.dead.letter队列。