RabbitMQ之延迟队列

消息中间件

发布日期: 2022-11-23

更新日期: 2022-11-24

文章字数: 2.5k

阅读时长: 12 分

延迟队列概念

延时队列,队列内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素是希望在指定时间到了以后或之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

延迟队列使用场景：

订单在十分钟之内未支付则自动取消

新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒

用户注册成功后，如果三天内没有登陆则进行短信提醒

用户发起退款，如果三天内没有得到处理则通知相关运营人员

预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议

TTL的两种设置方式

队列设置TTL

在创建队列的时候设置队列的 x-message-ttl 属性

Map<String, Object> params = new HashMap<>();
params.put("x-message-ttl",5000);
return QueueBuilder.durable("QA").withArguments(args).build(); // QA 队列的最大存活时间位 5000 毫秒

消息设置TTL

针对每条消息设置 TTL

rabbitTemplate.converAndSend("X","XC",message,correlationData -> {
    correlationData.getMessageProperties().setExpiration("5000");
});

区别

如果设置了队列的 TTL 属性，那么一旦消息过期，就会被队列丢弃(如果配置了死信队列被丢到死信队列中)，而第二种方式，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间，具体看下方案例。

另外，还需要注意的一点是，如果不设置 TTL，表示消息永远不会过期，如果将 TTL 设置为 0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢

案例

创建springboot工程

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.ssm</groupId>
    <artifactId>springboot-rabbitmq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-rabbitmq</name>
    <description>springboot-rabbitmq</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.47</version>
        </dependency>
        <!--swagger-->
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger2</artifactId>
            <version>2.9.2</version>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger-ui</artifactId>
            <version>2.9.2</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <!--RabbitMQ 依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

application.properties

server.port=8088
spring.rabbitmq.host=192.168.174.131
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=...59

创建两个队列 QA 和 QB，两个队列的 TTL 分别设置为 10S 和 40S，然后再创建一个交换机 X 和死信交换机 Y，它们的类型都是 direct，创建一个死信队列 QD，它们的绑定关系如下：

代码架构图

原先配置队列信息，写在了生产者和消费者代码中，现在可写在配置类中，生产者只发消息，消费者只接受消息

TtlQueueConfig



/**
 * @author shaoshao
 * @Date 2022/11/24 12:18
 * @Description: TTL配置类
 */
@Configuration
public class TtlQueueConfig {
    // 普通交换机
    public static final String X_EXCHANGE = "X";
    // 死信交换机
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    // 普通队列名称
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    //死信队列的名称
    public static final String DEAD_LETTER_QUEUE = "QD";

    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }

    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    @Bean("queueA")
    public Queue queueA() {
        HashMap<String, Object> arguments = new HashMap<>(3);
        //设置死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key", "YD");
        //设置TTL 10s 单位是ms
        arguments.put("x-message-ttl", 10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
    }

    //声明普通队列  要有ttl 为40s
    @Bean("queueB")
    public Queue queueB() {
        Map<String, Object> arguments = new HashMap<>(3);
        //设置死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key", "YD");
        //设置TTL 40s 单位是ms
        arguments.put("x-message-ttl", 40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
    }

    //声明死信队列
    @Bean("queueD")
    public Queue queueD() {
        return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
    }

    //声明队列 QA 绑定 X 交换机
    @Bean
    public Binding queueABindingX(@Qualifier("queueA") Queue queueA, @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueA).to(xExchange)
                .with("XA");
    }

    //声明队列 QB 绑定 X 交换机
    @Bean
    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB, @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueB).to(xExchange)
                .with("XB");
    }

    //声明队列 QD 绑定 Y 交换机
    @Bean
    public Binding queueDBindingY(@Qualifier("queueD") Queue queueD,
                                  @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }

}

SendMessageController



/**
 * @author shaoshao
 * @Date 2022/11/24 19:05
 * @Description:
 */
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMessageController {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * @Description: 发送消息
     * @Date: 2022/11/24 19:09
     * @Param: [message]
     * @Return: void
     */
    @GetMapping("/sendMsg/{message}")
    public void sendMsg(@PathVariable String message) {
        rabbitTemplate.convertAndSend("X", "XA", "来自消息ttl为10s的队列：" + message);
        rabbitTemplate.convertAndSend("X", "XB", "来自消息ttl为40s的队列：" + message);
        log.info("{},SendMessageController.sendMsg业务结束,结果是: {}", new Date().toString(), message);

    }
}

DeadLetterQueueConsumer



/**
 * @author shaoshao
 * @Date 2022/11/24 19:10
 * @Description: 队列TTL 消费者
 */
@Slf4j
@Component
public class DeadLetterQueueConsumer {

    @RabbitListener(queues = "QD")
    public void receiveD(Message message, Channel channel) throws Exception {
        String msg = new String(message.getBody());
        log.info("当前时间：{} ，收到死信队列消息：{}", new Date().toString(), msg);
    }

}

发送请求：http://localhost:8888/ttl/sendMsg/哈哈，观察控制台输出情况

测试成功

延时队列优化

在这里新增了一个队列 QC，该队列不设置 TTL 时间，根据前端的请求确定 TTL 时间，绑定关系如下：

配置类新增代码



/**
 * @author shaoshao
 * @Date 2022/11/24 12:18
 * @Description: TTL配置类
 */
@Configuration
public class TtlQueueConfig {
	...
        
    public static final String QUEUE_C = "QC";

	...
    //声明普通队列
    @Bean("queueC")
    public Queue queueC() {
        Map<String, Object> arguments = new HashMap<>(3);
        //设置死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key", "YD");
        return QueueBuilder.durable(QUEUE_C).withArguments(arguments).build();
    }
	...
    //声明队列 QC 绑定 X 交换机
    @Bean
    public Binding queueCBindingX(@Qualifier("queueC") Queue queueB, @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueB).to(xExchange)
                .with("XC");
    }
	...

}

SendMessageController新增代码



/**
 * @author shaoshao
 * @Date 2022/11/24 19:05
 * @Description:
 */
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMessageController {
    @Autowired
    private RabbitTemplate rabbitTemplate;

	...
        
    /**
     * @Description: 发送消息 设置TTL
     * @Date: 2022/11/24 20:17
     * @Param: [message, ttlTime]
     * @Return: void
     */
    @GetMapping("/sendExpirationMsg/{message}/{ttlTime}")
    public void sendMsg(@PathVariable String message, @PathVariable String ttlTime) {
        rabbitTemplate.convertAndSend("X", "XC", message, msg -> {
            // 设置发送消息时的延迟时长
            msg.getMessageProperties().setExpiration(ttlTime);
            return msg;
        });
        log.info("{},SendMessageController.sendMsg业务结束,发送一条  {}  msTTL信息给队列QC: {}", new Date().toString(), ttlTime, message);

    }
}

重启springboot
- http://localhost:8088/ttl/sendExpirationMsg/你好/20000
- http://localhost:8088/ttl/sendExpirationMsg/你好/2000

测试成功，出现问题

看起来似乎没什么问题，但是在最开始的时候，就介绍过如果使用在消息属性上设置 TTL 的方式，消息可能并不会按时「死亡」

问题：20s的先发送，随后发送2s的，但最后都是20s后才收到，也就是2s的排在了20s的后面，非得等到20s后才能执行2s的，而不是2s后收到

因为 RabbitMQ 只会检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行

Rabbitmq插件实现延迟队列

安装延时队列插件

官网：https://www.rabbitmq.com/community-plugins.html

安装过程请看docker那篇文章

打开 Web 界面，查看交换机的新增功能列表，如果多出了如图所示，代表成功添加插件

这里新增了一个队列 delayed.queue，一个自定义交换机 delayed.exchange，绑定关系如下:

DelayedQueueConfig


/**
 * @author shaoshao
 * @Date 2022/11/24 21:52
 * @Description: 延迟队列
 */
@Configuration
public class DelayedQueueConfig {
    //交换机
    public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
    //队列
    public static final String DELAYED_QUEUE_NAME = "delayed.queue";
    //routingKey
    public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";


    @Bean
    public Queue delayedQueue() {
        return new Queue(DELAYED_QUEUE_NAME);
    }

    //声明交换机,基于插件的交换机
    @Bean
    public CustomExchange delayedExchange() {

        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-delayed-type", "direct");
        /**
         * 1.交换机的名称
         * 2.交换机的类型 x-delayed-message
         * 3.是否需要持久化
         * 4.是否需要自动删除
         * 5.其他的参数
         */
        return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message",
                true, false, arguments);
    }

    //绑定
    @Bean
    public Binding delayedQueueBindingDelayedExchange(
            @Qualifier("delayedQueue") Queue delayedQueue,
            @Qualifier("delayedExchange") CustomExchange delayedExchange) {
        return BindingBuilder.bind(delayedQueue).to(delayedExchange)
                .with(DELAYED_ROUTING_KEY).noargs();
    }
}

SendMessageController


/**
 * @author shaoshao
 * @Date 2022/11/24 19:05
 * @Description: 生产者 发送消息
 */
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMessageController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    ...
        
    /**
     * @Description: 基于延迟插件 发送消息
     * @Date: 2022/11/24 22:11
     * @Param: [message, delayTime]
     * @Return: void
     */
    @GetMapping("/sendDelayMsg/{message}/{delayTime}")
    public void sendMsg(@PathVariable String message, @PathVariable Integer delayTime) {
        rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME, DelayedQueueConfig.DELAYED_ROUTING_KEY, message, msg -> {
            // 设置发送消息时的延迟时长
            msg.getMessageProperties().setDelay(delayTime);
            return msg;
        });
        log.info("{},SendMessageController.sendMsg业务结束,发送一条  {}  msTTL信息给队列QC: {}", new Date().toString(), delayTime, message);
    }
}

DelayQueueConsumer

package com.ssm.springbootrabbitmq.consumer;

import com.ssm.springbootrabbitmq.config.DelayedQueueConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author shaoshao
 * @Date 2022/11/24 22:11
 * @Description: 基于插件的延迟消息
 */
@Slf4j
@Component
public class DelayQueueConsumer {

    @RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
    public void receiveDelayQueue(Message message) {
        String msg = new String(message.getBody());
        log.info("当前时间：{},收到延时队列的消息：{}", new Date().toString(), msg);
    }


}

重启springboot访问地址：
- http://localhost:8088/ttl/sendDelayMsg/hello1/20000
- http://localhost:8088/ttl/sendDelayMsg/hello2/2000

验证成功，符合预期

可以看到哪怕 hello1 需要20秒再进入延时队列，hello2 2 秒后直接进入延时队列，无需等待 hello1

shaoshaossm

https://shaoshaossm.github.io/2022-11-23-rabbitmq-zhi-yan-chi-dui-lie.html

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 shaoshaossm !

RabbitMQ

ElasticSearch的基本操作

2022-11-25 搜索

ElasticSearch

RabbitMQ之死信队列

2022-11-23 消息中间件

RabbitMQ

TTL的两种设置方式

案例

延时队列优化

Rabbitmq插件实现延迟队列

你的赏识是我前进的动力