AI 助理
备案控制台
开发者社区 数据库 文章 正文

RabbitMQ延迟队列概念基础总结和实战

2021-12-31 389
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： RabbitMQ延迟队列概念基础总结和实战

1、延迟队列

1.1、概念

  • 延迟队列，队列内部是有序的。延迟队列中的元素是希望在指定时间到了之后或之前被取出和处理。简单来说，延迟队列就是用来存放需要在指定时间被处理的元素的队列。

1.2、使用场景

  1. 订单在十分钟之内未支付则自动取消。
  2. 新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
  3. 用户注册成功后，如果三天内没有登陆则进行消息提醒。
  4. 用户发起退款如果三天内没有得到处理则通知相关运营人员。
  5. 预定会议后，需要在预定的时间点前十分钟通知各个参会人员参加会议。
  • 这些场景都有一个特点，需要在某个事件发生之后，或者之前的指定时间点完成某一项任务。

1.3、SpringBoot整合

1. 引入的依赖

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <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>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.57</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <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.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
        </dependency>
    </dependencies>

2. 相关的配置

spring.rabbitmq.host=192.168.123.129
spring.rabbitmq.port=5672
spring.datasource.username=admin
spring.datasource.password=123

3. 添加Swagger配置类

package com.xiao.springbootrabbitmq.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.service.Contact;
import springfox.documentation.spi.DocumentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;

@Configuration
@EnableSwagger2
public class SwaggerConfig {

    @Bean
    public Docket webApiConfig(){
        return new Docket(DocumentationType.SWAGGER_2)
                .groupName("webApi")
                .apiInfo(webApiInfo())
                .select()
                .build();
    }

    private ApiInfo webApiInfo(){
        return new ApiInfoBuilder()
                .title("rabbitmq接口文档")
                .description("本文档描述了rabbitmq微服务接口定义")
                .version("1.0")
                .contact(new Contact("enjoy6288","http://atguigu.com","6382472874@qq.com"))
                .build();
    }
}

1.4、队列TTL

1.4.1、代码架构图

在这里插入图片描述

  • 创建两个队列QA和QB，两者队列TTL分别设置为10s和40s，然后创建一个死信队列QD。

1.4.2、TTL队列配置类

package com.xiao.springbootrabbitmq.config;

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

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

@Configuration
public class TtlQueueConfig {

    public static final String X_EXCHANGE = "X";

    public static final String Y_EXCHANGE = "Y";

    public static final String QUEUE_A = "QA";

    public static final String QUEUE_B = "QB";

    public static final String DEAD_LETTER_QUEUE_D = "QD";

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

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

    @Bean("queueA")
    public Queue queueA(){
        Map<String,Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange",Y_EXCHANGE);
        map.put("x-dead-letter-routing-key","YD");
        map.put("x-message-ttl",10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(map).build();
    }

    @Bean("queueB")
    public Queue queueB(){
        Map<String,Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange",Y_EXCHANGE);
        map.put("x-dead-letter-routing-key","YD");
        map.put("x-message-ttl",40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(map).build();
    }

    @Bean("queueD")
    public Queue queueD(){
        return QueueBuilder.durable(DEAD_LETTER_QUEUE_D).build();
    }

    @Bean
    public Binding QueueABindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }

    @Bean
    public Binding QueueBBindingX(@Qualifier("queueB") Queue queueB,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueB).to(xExchange).with("XB");
    }

    @Bean
    public Binding QueueDBindingY(@Qualifier("queueD") Queue queueD,
                                  @Qualifier("yExchange") DirectExchange yExchange){
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }
}

1.4.3、生产者代码

package com.xiao.springbootrabbitmq.controller;


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendMsg/{message}")
    public void sendMessage(@PathVariable String message){

        log.info("当前时间为：{}，当前发送的消息为：{}",new Date().toString(),message);
        rabbitTemplate.convertAndSend("X","XA","当前的消息来自ttl为10s的队列：" + message);
        rabbitTemplate.convertAndSend("X","XB","当前的消息来自ttl为40s的队列：" + message);
    }
}

1.4.4、消费者代码

package com.xiao.springbootrabbitmq.consumer;



import com.rabbitmq.client.Channel;
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;

@Slf4j
@Component
public class DeadLetterQueueConsumer {

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

}

1.4.5、测试结果

在这里插入图片描述

1.4.6、不足之处

  • 如果这样使用的话，那么每增加一个新的时间需求的话，就要新增一个队列。那么面对许多个需求就要新增许多个队列。

1.5、延迟队列优化

1.5.1、代码架构图

在这里插入图片描述

  • QC不设置延迟时间，其延迟时间由生产者进行设置，那么我们所形成的就是一个通用队列了。

1.5.2、配置类上添加的代码

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

    @Bean("queueC")
    public Queue queueC(){
        Map<String,Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange",Y_EXCHANGE);
        map.put("x-dead-letter-routing-key","YD");
        return QueueBuilder.durable(QUEUE_C).withArguments(map).build();
    }

    @Bean
    public Binding QueueCBindingX(@Qualifier("queueC") Queue queueC,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueC).to(xExchange).with("XC");
    }

1.5.3、生产者上添加的代码

    @GetMapping("sendExpirationMsg/{message}/{ttlTime}")
    public void sendMsg(@PathVariable String message,@PathVariable String ttlTime){
        rabbitTemplate.convertAndSend("X","XC",message,correlationDate->{
            correlationDate.getMessageProperties().setExpiration(ttlTime);
            return correlationDate;
        });
        log.info("当前时间为：{}，发送一条时长{}毫秒TTL的消息给队列C：{}",new Date().toString(),ttlTime,message);

    }

1.5.4、测试结果

在这里插入图片描述

1.5.5、存在的问题

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

1.6、Rabbitmq插件实现延迟队列

1.6.1、安装插件

  1. 下载地址

插件地址

  1. 将插件复制到/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins
  2. 执行rabbitmq-plugins enable rabbitmq_delayed_message_exchange
  3. 安装成功之后

在这里插入图片描述

1.6.2、代码架构图

在这里插入图片描述

1.6.3、配置类代码

@Configuration
    public class DelayedQueueConfig {
    public static final String DELAYED_QUEUE_NAME = "delayed.queue";
    public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
    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> args = new HashMap<>();
        //自定义交换机的类型
        args.put("x-delayed-type", "direct");
        return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false,args);
    }
    
    @Bean
    public Binding bindingDelayedQueue(@Qualifier("delayedQueue") Queue queue,
                                    @Qualifier("delayedExchange") CustomExchange delayedExchange) {
        return BindingBuilder.bind(queue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
    }
}

1.6.4、生产者代码

    public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
    public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
    
    @GetMapping("sendDelayMsg/{message}/{delayTime}")
    public void sendMsg(@PathVariable String message,@PathVariable Integer delayTime) {
    
        rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, message,correlationData ->{
            correlationData.getMessageProperties().setDelay(delayTime);
            return correlationData;
        });
        log.info(" 当 前 时 间 ： {}, 发 送 一 条 延 迟 {} 毫秒的信息给队列 delayed.queue:{}", new Date(),delayTime, message);

1.6.5、消费者代码

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

1.6.6、测试结果

在这里插入图片描述

  • 过期时间短的优先进入延时队列。

1.7、总结

  • 延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的利用RabbitMQ 的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过 RabbitMQ 集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。
  • 当然，延时队列还有很多其它选择，比如利用 Java 的 DelayQueue，利用Redis 的 zset ，利用 Quartz或者利用kafka 的时间轮 ，这些方式各有特点,看需要适用的场景
文章标签：
云消息队列 MQ
Java
消息中间件
NoSQL
Redis
Kafka
关键词：
云消息队列 MQ队列
云消息队列 MQ实战
云消息队列 MQ延迟队列
云消息队列 MQ延迟
云消息队列 MQ概念
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
崇尚学技术的科班人
目录
相关文章
何雨晨
|
1月前
|
消息中间件 存储 监控
RabbitMQ 队列之战：Classic 和 Quorum 的性能洞察
RabbitMQ 是一个功能强大的消息代理，用于分布式应用程序间的通信。它通过队列临时存储消息，支持异步通信和解耦。经典队列适合高吞吐量和低延迟场景，而仲裁队列则提供高可用性和容错能力，适用于关键任务系统。选择哪种队列取决于性能、持久性和容错性的需求。
何雨晨
135 6
bruce刘晓伟-18435
|
1月前
|
消息中间件 数据采集 中间件
RabbitMQ的使用—实战
RabbitMQ的使用—实战
bruce刘晓伟-18435
81 0
程序员小海绵
|
2月前
|
消息中间件 JSON Java
谷粒商城笔记+踩坑（22）——库存自动解锁。RabbitMQ延迟队列
库存自动解锁。RabbitMQ延迟队列
程序员小海绵
80 0
谷粒商城笔记+踩坑（22）——库存自动解锁。RabbitMQ延迟队列
游客j4mujezz7vm7y
|
2月前
|
消息中间件
rabbitmq,&队列
rabbitmq,&队列
游客j4mujezz7vm7y
35 0
喜欢猪猪
|
2月前
|
消息中间件 缓存 Java
RocketMQ的JAVA落地实战
RocketMQ作为一款高性能、高可靠、高实时、分布式特点的消息中间件，其核心作用主要体现在异步处理、削峰填谷以及系统解耦三个方面。
喜欢猪猪
175 0
小宋1021
|
2月前
|
消息中间件 JSON Java
玩转RabbitMQ声明队列交换机、消息转换器
玩转RabbitMQ声明队列交换机、消息转换器
小宋1021
87 0
土木林森
|
3月前
|
消息中间件 存储 NoSQL
MQ的顺序性保证：顺序队列、消息编号、分布式锁，一文全掌握！
【8月更文挑战第24天】消息队列（MQ）是分布式系统的关键组件，用于实现系统解耦、提升可扩展性和可用性。保证消息顺序性是其重要挑战之一。本文介绍三种常用策略：顺序队列、消息编号与分布式锁，通过示例展示如何确保消息按需排序。这些方法各有优势，可根据实际场景灵活选用。提供的Java示例有助于加深理解与实践应用。
土木林森
93 2
FeelTouchLabs
|
3月前
|
消息中间件 Kafka Apache
kafka vs rocketmq: 不要只顾着吞吐量而忘了延迟这个指标
这篇文章讨论了Apache RocketMQ和Kafka的对比，强调RocketMQ在低延迟、消息重试与追踪、海量Topic、多租户等方面进行了优化，特别是在小包非批量和大量分区场景下的吞吐量超越Kafka，适合电商和金融领域等高并发、高可靠和高可用场景。
FeelTouchLabs
106 0
embelfe_segge
|
消息中间件 Linux
centos7 yum快速安装rabbitmq服务
centos7 yum快速安装rabbitmq服务
embelfe_segge
227 0
杨宸杨
|
消息中间件 中间件 微服务
RabbitMQ 入门简介及安装
RabbitMQ 入门简介及安装
杨宸杨
122 0

热门文章

最新文章

  • 1
    RabbitMQ：四种ExChange用法
  • 2
    SpringBoot 整合 AMQP（RabbitMQ）
  • 3
    未读消息（小红点），前端与 RabbitMQ实时消息推送实践，贼简单~
  • 4
    SpringBoot整合RabbitMQ实现消息的发送与接收，确认消息，延时消息
  • 5
    rabbitmq开启web界面教程
  • 6
    安装配置rabbitmq
  • 7
    一些你会忘记的RabbitMQ集群和仲裁队列小知识点
  • 8
    3分钟实现SpringBoot集成RabbitMQ，实现消息队列服务！
  • 9
    rabbitmq逻辑与开发
  • 10
    转载RabbitMQ入门（3）--发布和订阅
  • 1
    win10 安装RabbitMQ的步骤--和报错解决
    89
  • 2
    在 Windows 平台搭建 MQTT 服务
    115
  • 3
    mq报错abbit@syld36: * connected to epmd (port 4369) on syld36 * epmd reports node ‘rabbit‘ uses po
    74
  • 4
    Spring Boot与RabbitMQ的集成
    89
  • 5
    Docker部署RabbitMQ消息中间件
    277
  • 6
    JAVA面试之MQ
    70
  • 7
    EMQX 与 MQTT: AI 大模型时代的分布式数据中枢
    240
  • 8
    如何在Java中使用RabbitMQ
    248
  • 9
    【RocketMQ系列十四】RocketMQ中消息堆积如何处理
    1378
  • 10
    【RocketMQ系列十三】RocketMQ的集群核心概念之消费重试&死信队列&幂等消息的出现以及处理
    131

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 快速体验云消息队列RocketMQ版
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 部署RabbitMQ
    • 下一篇
    手把手教你白嫖阿里云服务器(免费领服务器)