当前位置：首页 > article >正文

Rabbitmq--延迟消息

article 2025/3/11 10:32:19

13.延迟消息

延迟消息：生产者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才会收到消息

延迟任务：一定时间之后才会执行的任务

1.死信交换机

当一个队列中的某条消息满足下列情况之一时，就会成为死信（dead letter）：

消费者使用 basic.reject 或 basic.nack 声明消费失败，并且消息的 requeue 参数设置为 false
过期消息（达到了队列或消息本身设置的过期时间），消息超时后无人消费
要投递的队列消息堆积满了，最早的消息可能成为死信

如果队列通过 dead-letter-exchange 属性指定了一个交换机，那么该队列中的死信就会投递到这个交换机中，这个交换机称为死信交换机（Dead Letter Exchange，简称DLX）

利用死信交换机的特点，可以实现发送延迟消息的功能

2.延迟消息插件（推荐使用）

1.下载并安装延迟插件

RabbitMQ 的官方推出了一个插件，原生支持延迟消息功能。该插件的原理是设计了一种支持延迟消息功能的交换机，当消息投递到交换机后，可以将消息暂存一段时间，时间到了之后再将消息投递到队列中

插件的下载地址：rabbitmq-delayed-message-exchange

下载完插件后，运行以下指令，在输出信息中找到 Mounts ，再找到 RabbitMQ 的插件的安装目

sudo docker inspect rabbitmq

然后进入 RabbitMQ 的插件的安装目录，将刚才下载的插件上传到该目录下

sudo chmod +rx -R /var/lib/docker

//接着打开/var/lib/docker/volumes/rabbitmq-plugins/_data目录的写权限（如果修改权限不生效，请切换到 root 用户执行指令）
sudo chmod 777 /var/lib/docker/volumes/rabbitmq-plugins/_data

//将刚才下载的插件上传到/var/lib/docker/volumes/rabbitmq-plugins/_data目录
//上传成功后将/var/lib/docker/volumes/rabbitmq-plugins/_data目录的权限复原
sudo chmod 755 /var/lib/docker/volumes/rabbitmq-plugins/_data

//最后进入容器内部，运行指令安装插件，安装完成后退出容器内部
sudo docker exec -it rabbitmq bash
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
exit

看到以下信息，说明插件安装成功了

2.在 Java 代码中发送延迟消息

声明延迟交换机

@Bean
public DirectExchange delayExchange() {
    return ExchangeBuilder.directExchange("delay.direct").delayed().build();
}

声明队列和延迟交换机，并将队列和延迟交换机绑定在一起

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "delay.queue"),
        exchange = @Exchange(name = "delay.direct", delayed = "true", type = ExchangeTypes.DIRECT),
        key = "delay"
))
public void listenDelayQueue(String message) {
    SimpleDateFormat simpleDateFormat = new SimpleDateFormat();
    simpleDateFormat.applyPattern("yyyy-MM-dd HH:mm:ss SSS");
    System.out.println("消费者收到了 delay.queue的消息: " + message + "，时间：" + simpleDateFormat.format(System.currentTimeMillis()));
}

编写测试方法，测试发送延迟消息

@Test
void testSendDelayMessage() {
    rabbitTemplate.convertAndSend("delay.direct", "delay", "Hello, DelayQueue!", new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            message.getMessageProperties().setDelay(10000); // 毫秒
            return message;
        }
    });

    SimpleDateFormat simpleDateFormat = new SimpleDateFormat();
    simpleDateFormat.applyPattern("yyyy-MM-dd HH:mm:ss SSS");
    System.out.println("发送消息成功！发送时间：" + simpleDateFormat.format(System.currentTimeMillis()));
}

发送延迟消息的本质是在消息头属性中添加 x-delay 属性

3. 延迟消息的原理和缺点

RabbitMQ 的延迟消息是怎么实现的呢？RabbitMQ 会自动维护一个时钟，这个时钟每隔一秒就跳动一次，如果对时钟的精度要求比较高的，可能还要精确到毫秒，甚至纳秒
RabbitMQ 会为发送到交换机的每一条延迟消息创建一个时钟，时钟运行的过程中需要 CPU 不断地进行计算。发送到交换机的延迟消息数越多，RabbitMQ 需要维护的时钟就越多，对 CPU 的占用率就越高（Spring 提供的定时任务的原理也是类似）
定时任务属于 CPU 密集型任务，中间涉及到的计算过程对 CPU 来说压力是很大的，所以说，采用延迟消息会给服务器的 CPU 带来更大的压力。当交换机中有非常多的延迟消息时，对 CPU 的压力就会特别大

4.取消超时订单

设置 30 分钟后检测订单支付状态实现起来非常简单，但是存在两个问题

如果并发较高，30分钟可能堆积消息过多，对 MQ 压力很大
大多数订单在下单后 1 分钟内就会支付，但消息需要在 MQ 中等待30分钟，浪费资源

5.发送延迟检测订单的消息

我们定义一个实体类，用于记录延迟消息的内容和延迟消息的延迟时间列表（该实体类也是延迟消息的类型）

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class MultipleDelayMessage<T> {

    private T data;

    private List<Long> delayMillis;

    public MultipleDelayMessage() {

    }

    public MultipleDelayMessage(T data, Long... delayMillis) {
        this.data = data;
        this.delayMillis = new ArrayList<>(Arrays.asList(delayMillis));
    }

    public MultipleDelayMessage(T data, List<Long> delayMillis) {
        this.data = data;
        this.delayMillis = delayMillis;
    }

    public static <T> MultipleDelayMessage<T> of(T data, Long... delayMillis) {
        return new MultipleDelayMessage<>(data, new ArrayList<>(Arrays.asList(delayMillis)));
    }

    public static <T> MultipleDelayMessage<T> of(T data, List<Long> delayMillis) {
        return new MultipleDelayMessage<>(data, delayMillis);
    }

    public boolean hasNextDelay() {
        return !delayMillis.isEmpty();
    }

    public Long removeNextDelay() {
        return delayMillis.remove(0);
    }

    public T getData() {
        return data;
    }

    public void setData(T data) {
        this.data = data;
    }

    public List<Long> getDelayMillis() {
        return delayMillis;
    }

    public void setDelayMillis(List<Long> delayMillis) {
        this.delayMillis = delayMillis;
    }

    @Override
    public String toString() {
        return "MultipleDelayMessage{" +
                "data=" + data +
                ", delayMillis=" + delayMillis +
                '}';
    }

}

我们再定义一个发送延迟消息的消息处理器，供所有服务使用

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;

public class DelayMessagePostProcessor implements MessagePostProcessor {

    private final Integer delay;

    public DelayMessagePostProcessor(Integer delay) {
        this.delay = delay;
    }

    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        message.getMessageProperties().setDelay(delay);
        return message;
    }

}

改造后的发送延迟消息的测试方法

“delay.direct”：交换机的名称
“delay”：路由键（routing key），交换机会将消息发送到绑定到这个路由键的队列
“Hello, DelayQueue!”：实际要发送的消息内容
new DelayMessagePostProcessor(10000)：消息后处理器（Message Post Processor），用于在消息发送之前对消息进行修改

@Test
void testSendDelayMessage() {
    rabbitTemplate.convertAndSend("delay.direct", "delay", "Hello, DelayQueue!", new DelayMessagePostProcessor(10000));

    SimpleDateFormat simpleDateFormat = new SimpleDateFormat();
    simpleDateFormat.applyPattern("yyyy-MM-dd HH:mm:ss SSS");
    System.out.println("发送消息成功！发送时间：" + simpleDateFormat.format(System.currentTimeMillis()));
}

查看全文

http://www.kler.cn/a/579842.html