延迟队列的理解与使用

news2025/1/14 18:17:26

目录

一、场景引入

二、延迟队列的三种场景

1、TTL对队列进行延迟

2、创建通用延时消息对消息延迟

3、使用rabbitmq的延时队列插件

x-delayed-message使用

父pom文件

pom文件

配置文件

config

生产者

消费者

结果

一、场景引入

我们知道可以通过TTL来对队列进行设置过期时间；通过后置处理器MessagePostProcessor对消息进行设置过期时间；

 那么根据TTL及MessagePostProcessor机制可以处理关于延迟方面的问题。

比如：秒杀之后，给30分钟时间进行支付，如果30分钟后，没有支付，订单取消。

比如：餐厅订座,A用户早上8点预定的某家餐厅下午6点的座位,B用户中午12点预定的下午5点的座位；根据场景我们需要的时先让B用户进行消费,然后A用户再消费;这时TTL和MessagePostProcessor延迟就已经不能满足订餐的场景了；

因为TTL是对队列进行延迟,MessagePostProcessor是对消息进行延迟，但是MessagePostProcessor对消息延迟是不能根据订座的时间去排序消费的；

/**
 * 比如当我们发送第一个消息时延迟的时间时50s,而发送的第二个消息延迟时间是30s,虽然延迟30s的消息比延迟50s发送的晚
 * 但按照我们设想的情况,延迟30s的消息应该先消费;可是实际情况却不是这样,而是延迟50s的消息到达时间后 30s的才能消费!(队列先进先出)
 * 那这样此方式的不足就出现了!
 * 场景:
 *      A用户和B用户预定餐厅,A用户先开始预定的,预定的是下午6点。B用户比A用户预定操作晚一些,但是B用户预定的时间是下午5点。通过此场景
 *      我们希望的是B用户先进行用餐(因为他预定的吃饭时间比A早一些,需要先安排吃饭。不能说A用户没到6点B用户预定5点的吃不了。),根据此
 *      场景 之前的队列延迟还是消息延迟都不能满足场景需求了,这样就需要另一种延迟方式进行解决了! ->使用rabbitmq的延时队列插件
 */

/**
 * 注意：
 *      TTL是对队列进行延迟,只要是在此队列中的消息都会按照TTL设置时间进行延迟;
 *      MessagePostProcessor是对消息进行延迟;
 *
 *      如果我们不仅使用了消息延迟,而且还使用了队列延迟,那么延迟的时间就会以小的时间为准!
 *   理解:
 *      如果a消息设置的消息延迟是30s,b消息设置的延迟时间是90s,队列设置的延迟是60s。那么a消息最终的延迟是30s(a的消息延迟与队列延迟
 *      比对以延迟时间小的为准！),b消息最终延迟的时间是60s(b的消息延迟与队列延迟比对以延迟的时间小的为准！)
 */

 二、延迟队列的三种场景

1、TTL对队列进行延迟

   @Bean
    public Queue directQueueLong(){
        return   QueueBuilder.durable("业务队列名称")
                .deadLetterExchange("死信交换机名称")
                .deadLetterRoutingKey("死信队列 RoutingKey")
                .ttl(20000) // 消息停留时间
                //.maxLength(500)
                .build();
    }

监听死信队列，即可处理超时的消息队列

缺点：

上述实现方式中，ttl延时队列中所有的消息超时时间都是一样的，如果不同消息想设置不一样的超时时间，就需要建立多个不同超时时间的消息队列，比较麻烦，且不利于维护。

 2、创建通用延时消息对消息延迟

rabbitTemplate.convertAndSend("交换机名称", "RoutingKey","对象",
message => {

 
        message.getMessageProperties().setExpiration(String.valueOf(5000))
       // 设置消息的持久化属性
 
//这样发送的消息就会被持久化到 RabbitMQ 中，即使 RabbitMQ 重启，消息也不会丢失。                        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);

        return message;
            }
           );

缺点：

该种方式可以创建一个承载不同超时时间消息的消息队列，但是这种方式有一个问题，如果消息队列中排在前面的消息没有到超时时间，即使后面的消息到了超时时间，先到超时时间的消息也不会进入死信队列，而是先检查排在最前面的消息队列是否到了超时时间，如果到了超时时间才会继续检查后面的消息。

 3、使用rabbitmq的延时队列插件

使用rabbitmq的延时队列插件，实现同一个队列中有多个不同超时时间的消息，并按时间超时顺序出队

1、下载延迟插件

在 RabbitMQ 的 3.5.7 版本之后，提供了一个插件（rabbitmq-delayed-message-exchange）来实现延迟队列 ，同时需保证 Erlang/OPT 版本为 18.0 之后。

下载地址：

https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

2、安装插件并启用

下载完成后直接把插件放在 /home/rabbitmq 目录，然后拷贝到容器内plugins目录下（my-rabbit是容器的name，也可以使用容器id）

docker cp /home/rabbitmq/rabbitmq_delayed_message_exchange-3.12.0.ez rabbitmq:/plugins

 进入 Docker 容器

docker exec -it rabbit /bin/bash

 在plugins内启用插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

 退出容器

exit

重启 RabbitMQ

docker restart my-rabbit

 貌似不重启也能生效！

 结果：

 就多了一个x-delayed-message交换机

x-delayed-message使用

父pom文件

<?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.7.1</version>
<!--        <version>2.2.5.RELEASE</version>-->
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.chensir</groupId>
    <artifactId>spring-boot-rabbitmq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>spring-boot-rabbitmq</name>

    <properties>
        <java.version>8</java.version>
        <hutool.version>5.8.3</hutool.version>
        <lombok.version>1.18.24</lombok.version>
    </properties>

    <description>spring-boot-rabbitmq</description>

    <packaging>pom</packaging>

    <modules>
        <module>direct-exchange</module>
        <module>fanout-exchange</module>
        <module>topic-exchange</module>
        <module>game-exchange</module>
        <module>dead-letter-queue</module>
        <module>delay-queue</module>
        <module>delay-queue2</module>
    </modules>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>cn.hutool</groupId>
                <artifactId>hutool-all</artifactId>
                <version>${hutool.version}</version>
            </dependency>

            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>${lombok.version}</version>
            </dependency>
        </dependencies>
    </dependencyManagement>


</project>

pom文件

<?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>com.chensir</groupId>
        <artifactId>spring-boot-rabbitmq</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <relativePath>../pom.xml </relativePath>
    </parent>

    <artifactId>delay-queue2</artifactId>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

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

</project>

配置文件

logging.level.com.chensir = debug
server.port=8086
#host
spring.rabbitmq.host=121.40.100.66
#默认5672
spring.rabbitmq.port=5672
#用户名
spring.rabbitmq.username=guest
#密码
spring.rabbitmq.password=guest
#连接到代理时用的虚拟主机
spring.rabbitmq.virtual-host=/
#每个消费者每次可最大处理的nack消息数量
spring.rabbitmq.listener.simple.prefetch=1
#表示消息确认方式，其有三种配置方式，分别是none、manual(手动)和auto(自动)；默认auto
spring.rabbitmq.listener.simple.acknowledge-mode=auto
#监听重试是否可用
spring.rabbitmq.listener.simple.retry.enabled=true
#最大重试次数
spring.rabbitmq.listener.simple.retry.max-attempts=5
#最大重试时间间隔
spring.rabbitmq.listener.simple.retry.max-interval=3000ms
#第一次和第二次尝试传递消息的时间间隔
spring.rabbitmq.listener.simple.retry.initial-interval=1000ms
#应用于上一重试间隔的乘数
spring.rabbitmq.listener.simple.retry.multiplier=2
#决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
spring.rabbitmq.listener.simple.default-requeue-rejected=false

config

package com.chensir.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitConfig {

    @Bean
    public MessageConverter messageConverter(){
        return  new Jackson2JsonMessageConverter();
    }


    @Bean
    public CustomExchange customExchange(){
        Map<String,Object> args = new HashMap<>();
        //延迟时以direct直连方式绑定
        args.put("x-delayed-type","direct");

        // name:交换机名称 type:类型 固定值x-delayed-message
        return new CustomExchange("chen-x-delayedExchange","x-delayed-message",true,false,args);
    }
    @Bean
    public Queue directQueueLong(){
        return   QueueBuilder.durable("chen-DirectQueue")
                .build();
    }

    @Bean
    public Binding binding(){
        return  BindingBuilder.bind(directQueueLong()).to(customExchange()).with("direct123").noargs();
    }
}

生产者

package com.chensir.provider;

import com.chensir.model.OrderIngOk;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
@Slf4j
public class DirectProvider {

    @Resource
    private RabbitTemplate rabbitTemplate;

    public  void  send(){
            OrderIngOk orderIngOk = new OrderIngOk();
            orderIngOk.setOrderNo("202207149687-1");
            orderIngOk.setId(1);
            orderIngOk.setUserName("倪海杉前-延迟40秒");
            rabbitTemplate.convertAndSend("chen-x-delayedExchange", "direct123",orderIngOk,m->{
                m.getMessageProperties().setDelay(40*1000); //设置延迟时间，对延迟交换机有效
                // m.getMessageProperties().setExpiration(String.valueOf(30*1000)); 设置过期时间，对队列有效
                return m;
            });

        OrderIngOk orderIngOk2 = new OrderIngOk();
        orderIngOk2.setOrderNo("202207149687-2");
        orderIngOk2.setId(2);
        orderIngOk2.setUserName("倪海杉后-延迟30秒");
        rabbitTemplate.convertAndSend("chen-x-delayedExchange", "direct123",orderIngOk2,m->{
            m.getMessageProperties().setDelay(30*1000);
            return m;
        });
            log.debug("消息生产完成");



    }

}

消费者

package com.chensir.consumer;

import cn.hutool.json.JSONUtil;
import com.chensir.model.OrderIngOk;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
@Slf4j
public class DirectConsumer {

    @RabbitHandler
    @RabbitListener(queues = "chen-DirectQueue" )
    public void  process(OrderIngOk orderIngOk) throws IOException {
        try {
           // 处理业务开始
            log.debug("接受到消息,并正常处理结束,{}", JSONUtil.toJsonStr(orderIngOk));
            // 处理业务结束
        } catch (Exception ex){
            throw ex;
        }
    }
}

结果

2023-08-29 18:27:45.983 DEBUG 15568 --- [           main] com.chensir.provider.DirectProvider      : 消息生产完成
2023-08-29 18:28:16.143 DEBUG 15568 --- [ntContainer#0-1] com.chensir.consumer.DirectConsumer      : 接受到消息,并正常处理结束,{"id":2,"OrderNo":"202207149687-2","userName":"倪海杉后-延迟30秒"}
2023-08-29 18:28:26.057 DEBUG 15568 --- [ntContainer#0-1] com.chensir.consumer.DirectConsumer      : 接受到消息,并正常处理结束,{"id":1,"OrderNo":"202207149687-1","userName":"倪海杉前-延迟40秒"}

可见 延迟40s的是先发送的，但是最终结果是先消费延迟30s的。这样就能达到我们订座的场景需求了。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/944783.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

百数应用中心——生产制造管理解决方案解决行业难题

百数应用中心——生产制造管理解决方案解决行业难题

传统生产制造业面临着许多挑战&#xff0c;其中一些主要问题包括效率低下、交期压力大、需求预测不准确、生产模式复杂、异常响应慢、库存高和计划脱节等。这些问题不仅影响了生产效率和质量&#xff0c;也导致了不必要的成本和客户满意度下降。 生产制造管理应用对于企业的生产…
阅读更多...
[已解决] wget命令出现Unable to establish SSL connection.错误

[已解决] wget命令出现Unable to establish SSL connection.错误

问题 从win11上下载best.ckpt包时 遇到&#xff1a; Unable to establish SSL connection.错误 解决方案&#xff1a; 加上参数&#xff1a; 加上跳过验证证书的参数--no-check-certificate 有些网站不允许通过非浏览器的方式进行下载&#xff0c;使用代理既可以解决问题&am…
阅读更多...
MES管理系统中常用的数据模型有哪些

MES管理系统中常用的数据模型有哪些

在MES管理系统项目中&#xff0c;数据建模对于生产过程的监控、分析和管理具有至关重要的作用。本文将介绍一些常见的MES管理系统数据建模方面&#xff0c;并阐述它们在生产过程中的重要性和应用。 1、产品数据模型是MES系统中的基础模块之一。它涵盖了产品的基本信息、规格和属…
阅读更多...
JavaScript 中的箭头函数：如何使用简洁的语法

JavaScript 中的箭头函数：如何使用简洁的语法

了解有关 JavaScript 箭头函数的所有信息。我们将向您展示如何使用 ES6 箭头语法&#xff0c;以及在代码中利用箭头函数时需要注意的一些常见错误。您将看到许多示例来说明它们的工作原理。 JavaScript 箭头函数随着 ECMAScript 2015&#xff08;也称为 ES6&#xff09;的发布…
阅读更多...
【pyqt5界面化开发-4】垂直布局/水平布局+‘套娃‘布局

【pyqt5界面化开发-4】垂直布局/水平布局+‘套娃‘布局

目录 一、垂直布局 二、布局器的组合 三、水平布局垂直布局&#xff08;套娃&#xff09; 一、垂直布局 需要模块&#xff1a;QVBoxLayout # 垂直布局layout QVBoxLayout()………………# 应用设置的布局器self.setLayout(layout) 模块间的伸缩器&#xff08;可以理解为弹簧…
阅读更多...
InfluxDB在Windows的下载，安装，自启动

InfluxDB在Windows的下载，安装，自启动

1.官网下载 官网下载地址&#xff0c;我一般都喜欢官网下载&#xff0c;包含windows&#xff0c;linux&#xff0c;macos&#xff0c;docker&#xff0c;K8s甚至树莓派系统等。 2.启动 进入到文件夹下&#xff0c;双击启动exe的话会一闪而过&#xff0c;它告诉你要cmd用命令…
阅读更多...
uni-app中使用iconfont彩色图标

uni-app中使用iconfont彩色图标

uni-app中使用iconfont彩色图标 大家好&#xff0c;今天我们来学习一下uni-app中使用iconfont彩色图标&#xff0c;好好看&#xff0c;好好学&#xff0c;超详细的 第一步 首先&#xff0c;从iconfont官网&#xff08;iconfont-阿里巴巴矢量图标库&#xff09;选择自己需要的图…
阅读更多...
九号公司董事长高禄峰：立足科技创新，助力行业发展

九号公司董事长高禄峰：立足科技创新，助力行业发展

从整个电动两轮车发展时间线来看&#xff0c;尽管九号公司2019年才正式进入局电动两轮车市场&#xff0c;但公司凭借智能化入局&#xff0c;在短短三年时间里&#xff0c;已经成长为智能电动两轮车市场的领军者&#xff0c;在这期间&#xff0c;九号公司既没有多年的市场积累&a…
阅读更多...
基于 OV5640 摄像头理论知识讲解-典型工作模式配置

基于 OV5640 摄像头理论知识讲解-典型工作模式配置

基于OV2640/ OV5640 的图像采集显示系统系列文章目录&#xff1a; &#xff08;1&#xff09;基于 OV5640 摄像头理论知识讲解-成像和采样原理 &#xff08;2&#xff09;基于 OV5640 摄像头理论知识讲解-数字接口和控制接口 &#xff08;3&#xff09;基于 OV5640 摄像头理论知…
阅读更多...
单链表oj题

单链表oj题

&#x1f35f;1.反转链表 链接: https://leetcode.cn/problems/reverse-linked-list/description/ 思路一&#xff1a;遍历一遍的同时两两逆置 写法一&#xff1a; /*** Definition for singly-linked list.* struct ListNode {* int val;* struct ListNode *next…
阅读更多...
算法设计 || 第8题：用你的学号作为X； Y= ”20220528“ ,用列表法求出最长公共子序列的长度并给出所有的最长公共子序列(写出计算表格和答案)。

算法设计 || 第8题：用你的学号作为X； Y= ”20220528“ ,用列表法求出最长公共子序列的长度并给出所有的最长公共子序列(写出计算表格和答案)。

X2100300805 Y20220528 最长公共子序列&#xff1a;不需要连续 答案&#xff1a;2005或2008
阅读更多...
生成式AI背景下编程工作者的技术和高级软考理论的演进融合之路

生成式AI背景下编程工作者的技术和高级软考理论的演进融合之路

思考背景 近两次软考&#xff0c;我都参与了&#xff0c;2022年11月参加的是系统架构师的考试&#xff0c;2023年5月参加的是系统分析师的考试&#xff0c;去年参加系统架构是考试是完全的裸考和第一次考&#xff0c;成绩是选择题39&#xff0c;综合题46和论文48分&#xff0c…
阅读更多...
开发卡牌gamefi游戏需要多少钱？

开发卡牌gamefi游戏需要多少钱？

卡牌游戏作为一种受欢迎的游戏形式&#xff0c;吸引了众多开发者的关注。然而&#xff0c;开发一款成功的卡牌游戏需要全面考虑多个方面的因素&#xff0c;其中之一就是资金投入。本文将从专业性和投入回报的角度&#xff0c;探讨开发一款卡牌游戏所需的资金投入。 一、专业性的…
阅读更多...
[Agent]开发---csvAgent开发

[Agent]开发---csvAgent开发

资料&#xff1a; https://python.langchain.com/docs/integrations/toolkits/csvcsvagent需要csv文件和模型&#xff0c;该方法在后台使用pandas方法去处理该文件&#xff0c;然后调用LLM去回答问题 我们使用内置函数去调用csvagent。 AgentType根据自己需要下载&#xff1b;…
阅读更多...
国家公派必读|CSC各类留学人员奖学金资助和艰苦地区补贴标准

国家公派必读|CSC各类留学人员奖学金资助和艰苦地区补贴标准

国家留学基金委&#xff08;以下简称CSC&#xff09;为公派留学人员&#xff08;含高级研究人员、访问学者、博士后及留学生&#xff09;提供一定数额的奖学金&#xff0c;对艰苦地区还有特殊生活补贴&#xff0c;本文知识人网小编根据相关文件加以整理后撰文如下。 根据以上文…
阅读更多...
打破对ChatGPT的依赖以及如何应对ChatGPT的错误和幻觉

打破对ChatGPT的依赖以及如何应对ChatGPT的错误和幻觉

​ OpenAI的ChatGPT是第一个真正流行的生成式AI工具&#xff0c;但它可能不是最好的。现在是时候扩大你的AI视野了。 ChatGPT成为了基于大语言模型(LLM)的聊天机器人的同义词。但是现在是时候停止对ChatGPT的痴迷&#xff0c;开始发现这个新世界中强大的替代品了。 首先&a…
阅读更多...
StringBuilder类分享（1）

StringBuilder类分享（1）

一、StringBuilder说明 StringBuilder是一个可变的字符序列。这个类提供了一个与StringBuffer兼容的API&#xff0c;但不保证同步&#xff0c;即StringBuilder不是线程安全的&#xff0c;而StringBuffer是线程安全的。显然&#xff0c;StringBuilder要运行的更快一点。 这个类…
阅读更多...
编程题四大算法思想（一）——分治法：最大子数组问题、矩阵乘法的Strassen算法、最近点对问题、凸包问题

编程题四大算法思想（一）——分治法：最大子数组问题、矩阵乘法的Strassen算法、最近点对问题、凸包问题

文章目录 分治法分治策略分治算法的效率分析归并排序 具体应用问题最大子数组问题蛮力法分治法找跨越中间位置的最大子数组时间复杂度 矩阵乘法蛮力算法分治法Strassen矩阵乘法 分治法 方法 分治策略分治法效率分析——迭代法&#xff08;递归树法&#xff09;分治法效率分析—…
阅读更多...
学习JAVA打卡第四十六天

学习JAVA打卡第四十六天

Date和Calendar类 Date类 ⑴使用无参数构造方法 使用Date 类的无参数构造方法创建的对象可以获取本机的当前日期和时间&#xff0c;例如&#xff1a; Date nowtime new Date&#xff08;&#xff09;&#xff1b; ⑵使用带参数的构造方法 计算机系统将其自身的时间的设“…
阅读更多...
redis实战-项目集成git及redis实现短信验证码登录

redis实战-项目集成git及redis实现短信验证码登录

目录 IDEA集成git 传统session存在的问题 redis方案 业务流程 选用的数据结构 整体访问流程 发送短信验证码 获取校验验证码 配置登录拦截器 拦截器注册配置类 拦截器 用户状态刷新问题 刷新问题解决方案 IDEA集成git 远程仓库采用码云&#xff0c;创建好仓库&…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023