RabbitMQ延迟消息问题(含Demo工程)

2022-11-07 13:40:09 浏览数 (1)

目录

1、什么是死信交换机

2、TTL

2.1 Demo

1、准备接收超时死信的死信交换机

2、声明一个队列，并且指定TTL

3、发送消息

4、小结

3、延迟队列

3.1 安装DelayExchange插件

1、上传插件

2、安装插件

3.2 DelayExchange原理

3.3 使用DelayExchange

3.4 发送消息

3.5 小结

4、项目Demo地址

本篇博客带大家研究MQ的延迟消息问题，在此之前先了解一下死信交换机。

1、什么是死信交换机

首先我们要知道什么是死信？

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

  1. 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false。
  2. 消息是一个过期消息，超时无人消费。
  3. 要投递的队列消息堆积满了，最早的消息可能成为死信。

一般呢？一旦消息变成死信是会被我们丢弃的，但是有了死信交换机就不一样了。

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

其实呢，所谓的死信交换机就是一个普通交换机，只不过是某个队列用dead-letter-exchange这个属性绑定到一起了，当这个队列出现了死信，就会丢到我们这个死信交换机里了，就有点像垃圾桶一样的了。

如图，一个消息被消费者拒绝了，变成了死信：

因为simple.queue绑定了死信交换机 dl.direct，因此死信会投递给这个交换机：

 如果这个死信交换机也绑定了一个队列，则消息最终会进入这个存放死信的队列：

另外，队列将死信投递给死信交换机时，必须知道两个信息：

  • 死信交换机名称
  • 死信交换机与死信队列绑定的RoutingKey

这样才能确保投递的消息能到达死信交换机，并且正确的路由到死信队列。

小结：

什么样的消息会成为死信？

  1. 消息被消费者reject或者返回nack。
  2. 消息超时未消费。
  3. 队列满了。

2、TTL

TTL，也就是Time-To-Live。如果一个队列中的消息TTL结束仍未消费，则会变为死信，TTL超时分为两种情况：

  • 消息所在的队列设置了超时时间
  • 消息本身设置了超时时间

2.1 Demo

1、准备接收超时死信的死信交换机

在consumer服务的SpringRabbitListener中，定义一个新的消费者，并且声明 死信交换机、死信队列：

代码语言：javascript复制
 /**
   * @description:注解方式声明死信交换机、死信队列
   * @author: jie 
   * @time: 2022/3/5 10:30
  */
@RabbitListener(bindings = @QueueBinding(
            //队列，持久化为true
            value = @Queue(name = "dl.ttl.queue", durable = "true"),
            //交换机
            exchange = @Exchange(name = "dl.ttl.direct"),
            //Routing Key
            key = "ttl"
    ))
public void listenDlQueue(String msg){
        log.info("接收到 dl.ttl.queue的延迟消息：{}", msg);
}

2、声明一个队列，并且指定TTL

代码语言：javascript复制
package com.jie.mq.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class TTLMessageConfig  {
    /**
     * @description:交换机
     * @author: jie
     * @time: 2022/3/5 10:36
     */
    @Bean
    public DirectExchange ttlDirectExchange(){
        return new DirectExchange("ttl.direct");
    }
    /**
     * @description:队列
     * @author: jie
     * @time: 2022/3/5 10:38
     */
    @Bean
    public Queue ttlQueue(){
        return QueueBuilder
                //指定队列名称，并持久化
                .durable("ttl.queue")
                //设置队列的超时时间，10秒
                .ttl(10000)
                //指定死信交换机
                .deadLetterExchange("dl.ttl.direct")
                //设置RoutingKey
                .deadLetterRoutingKey("dl")
                .build();
    }
    /**
     * @description:将队列和交换机绑定
     * @author: jie
     * @time: 2022/3/5 10:41
     */
    @Bean
    public Binding ttlBinding(){
        return BindingBuilder.bind(ttlQueue()).to(ttlDirectExchange()).with("ttl");
    }
}

3、发送消息

3.1 不指定TTL。

代码语言：javascript复制
@Test
public void testTTLQueue() {
        // 创建消息
        String message = "hello, ttl queue";
        // 消息ID，需要封装到CorrelationData中
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        // 发送消息
        rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
        // 记录日志
        log.error("发送消息成功");
    }

消息发送时间：

 接受消息的时间：

因为队列的TTL值是10000ms，也就是10秒。可以看到消息发送与接收之间的时差刚好是10秒。

我们这个是基于队列去设置延迟时间，我们给队列设置了10秒钟，我们也可以给消息设置延迟。

3.2 指定TTL

代码语言：javascript复制
public void testTTLMessage() {
        //准备消息
        Message message = MessageBuilder
                .withBody("hell,TTL".getBytes(StandardCharsets.UTF_8))
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                //设置延迟时间
                .setExpiration("5000")
                .build();
        // 2.发送消息
        rabbitTemplate.convertAndSend("ttl.direct","ttl",message);
        //3、记录日志
        log.info("消息已经成功发送!");
    }

这里呢？我们的队列是10秒，而消息是5秒，到底是哪个优先呢？还是15秒呢？

消息发送时间：

 消息接受时间：

这里我们可以看出，当两个都有延迟的时候，它会以较短的时间为准。

4、小结

消息超时的两种方式是？

  • 给队列设置ttl属性，进入队列后超过ttl时间的消息变为死信
  • 给消息设置ttl属性，队列接收到消息超过ttl时间后变为死信

如何实现发送一个消息20秒后消费者才收到消息？

  • 给消息的目标队列指定死信交换机
  • 将消费者监听的队列绑定到死信交换机
  • 发送消息时给消息设置超时时间为20秒

3、延迟队列

概念：

利用TTL结合死信交换机，我们实现了消息发出后，消费者延迟收到消息的效果。这种消息模式就称为延迟队列（Delay Queue）模式。

延迟队列的使用场景包括：

1、延迟发送短信。 2、用户下单，如果用户在15 分钟内未支付，则自动取消。 3、预约工作会议，20分钟后自动通知所有参会人员。

因为延迟队列的需求非常多，所以RabbitMQ的官方也推出了一个插件，原生支持延迟队列效果。

这个插件就是DelayExchange插件。参考RabbitMQ的插件列表页面：Community Plugins — RabbitMQ

使用方式可以参考官网地址：Scheduling Messages with RabbitMQ | RabbitMQ - Blog

3.1 安装DelayExchange插件

官方的安装指南地址为：Scheduling Messages with RabbitMQ | RabbitMQ - Blog

上述文档是基于linux原生安装RabbitMQ，然后安装插件。

因为我是基于Docker安装RabbitMQ,所以下面我会讲解基于Docker来安装RabbitMQ插件。

RabbitMQ有一个官方的插件社区，地址为：Community Plugins — RabbitMQ

其中包含各种各样的插件，包括我们要使用的DelayExchange插件：

 下载好后，就会获得一个ez文件。

1、上传插件

因为我们是基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。

我之前设定的RabbitMQ的数据卷名称为mq-plugins，所以我使用下面命令查看数据卷：

代码语言：javascript复制
docker volume inspect mq-plugins

可以得到下面结果：

 接下来，将插件上传到这个目录即可：

2、安装插件

最后就是安装了，需要进入MQ容器内部来执行安装。我的容器名为mq，所以执行下面命令：

代码语言：javascript复制
docker exec -it mq bash

执行时，请将其中的 -it 后面的mq替换为你自己的容器名.

进入容器内部后，执行下面命令开启插件：

代码语言：javascript复制
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

3.2 DelayExchange原理

DelayExchange的本质还是官方的三种交换机，只是添加了延迟功能。因此使用时只需要声明一个交换机，交换机的类型可以是任意类型，然后设定delayed属性为true即可。

  1. 接收消息。
  2. 判断消息是否具备x-delay属性。
  3. 如果有x-delay属性，说明是延迟消息，持久化到硬盘，读取x-delay值，作为延迟时间。
  4. 返回routing not found结果给消息发送者。
  5. x-delay时间到期后，重新投递消息到指定队列。

3.3 使用DelayExchange

1、基于注解的方式（推荐）

 2、基于Bean的方式

3.4 发送消息

发送消息时，一定要携带x-delay属性，指定延迟的时间：

 发送消息时间：

 接受消息时间：

相差五秒，说明是有用的。

3.5 小结

延迟队列插件的使用步骤包括哪些？

•声明一个交换机，添加delayed属性为true •发送消息时，添加x-delay头，值为超时时间

4、项目Demo地址

无聊的英杰/RabbitMQ延迟消息问题

rabbitmq 容器镜像服务

0 人点赞