文章详情

登录

短信预约-IT技能 免费直播动态提醒

选择考试省份

  • 北京
  • 上海
  • 天津
  • 重庆
  • 河北
  • 山东
  • 辽宁
  • 黑龙江
  • 吉林
  • 甘肃
  • 青海
  • 河南
  • 江苏
  • 湖北
  • 湖南
  • 江西
  • 浙江
  • 广东
  • 云南
  • 福建
  • 海南
  • 山西
  • 四川
  • 陕西
  • 贵州
  • 安徽
  • 广西
  • 内蒙
  • 西藏
  • 新疆
  • 宁夏
  • 兵团

请输入下面的图形验证码

提交验证

短信预约提醒成功

详解RabbitMQ延迟队列的基本使用和优化

2023-05-19 08:58

关注

1.延迟队列基本介绍

一般队列中的元素总是希望能够早点被取出来进行处理，但是延迟队列中的元素则是希望可以在指定时间内被取出和处理，延迟队列中的元素都是带有时间属性的。延迟队列就是用来存放需要在指定时间被处理的元素的队列

在这里插入图片描述

延迟队列就是想要消息延迟一段时间后被处理，TTL可以让消息在延迟一段时间后变成死信。变成死信的消息都会被投递到死信队列中，这样的话，只要消费者一直消费死信队列里面的消息就可以了，因为里面的消息都是希望被马上处理的消息 生产者生产一条延时消息，根据需要延时时间的不同，通过不同的routing key把消息路由到不同的延迟队列，每一个队列都设置了不同的TTL属性，并且绑定在同一个死信交换机中，消息过期了以后，根据routing key的不同，又会被路由到不同的死信队列中，消费者只需要监听对应的死信队列进行处理就可以了。注意:不要造成重复消费

2.延迟队列使用场景

下面的场景需要使用延迟队列

  1. 订单在十分钟内没有支付就自动取消
  2. 新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒
  3. 账单在一周内没有支付，就会自动结算
  4. 用户注册成功以后，如果三天内没有登录就进行短信题提醒
  5. 用户发起退款，如果三天内没有得到处理则通知相关运营人员。
  6. 预定会议以后，需要提前十分钟通知各个参会人员参加会议。

3.Spring Boot集成RabbitMQ

3.1创建项目，引入依赖

在这里插入图片描述

相关依赖

 <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>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </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>

3.2application.properties配置文件

# RabbitMQ/配置
#服务器地址
spring.rabbitmq.host=服务器地址
#服务端口号
spring.rabbitmq.port=5672
#虚拟主机名称
spring.rabbitmq.virtual-host=/myhost
#用户名
spring.rabbitmq.username=admin
#密码
spring.rabbitmq.password=123456

3.3 队列TTL-代码结构图

在这里插入图片描述

3.4MQ配置类

package com.zyh.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 TtlQueueConfiguration {
    //普通交换机
    public static final String X_EXCHANGE = "X";
    //普通队列
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    //死信交换机
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    //死信队列QD
    public static final String QUEUE_D = "QD";
    
    @Bean
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }
    
    @Bean
    public Queue queueA() {
        //创建集合保存队列属性
        Map<String, Object> map = new HashMap<>();
        //设置该队列绑定的死信交换机名称
        map.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //设置routing key
        map.put("x-dead-letter-routing-key", "YD");
        //设置队列延迟时间 10秒
        map.put("x-message-ttl", 10000);
        //创建队列
        return QueueBuilder.durable(QUEUE_A).withArguments(map).build();
    }
    
    @Bean
    public Binding queueA_BindingX(@Qualifier("queueA") Queue queue, @Qualifier("xExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("XA");
    }
    
    @Bean
    public Queue queueB() {
        //创建集合保存队列属性
        Map<String, Object> map = new HashMap<>();
        //设置该队列绑定的死信交换机名称
        map.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //设置routing key
        map.put("x-dead-letter-routing-key", "YD");
        //设置队列延迟时间 10秒
        map.put("x-message-ttl", 40000);
        //创建队列
        return QueueBuilder.durable(QUEUE_A).withArguments(map).build();
    }
    
    @Bean
    public Binding queueB_BindingX(@Qualifier("queueB") Queue queue, @Qualifier("xExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("XB");
    }
    
    @Bean
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }
    
    @Bean
    public Queue queueD() {
        return new Queue(QUEUE_D);
    }
    
    @Bean
    public Binding deadLetterBindingQD(@Qualifier("queueD") Queue queue, @Qualifier("yExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("YD");
    }
}

3.5生产者代码

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMessageController {
    @Resource
    private RabbitTemplate rabbitTemplate;
    
    @GetMapping("/sendMessage/{message}")
    public void sendMessage(@PathVariable String message){
        //记录日志
        log.info("当前时间:{},发送一条信息给两个TTL队列:{}",new Date(),message);
        //给QA队列发送消息
        rabbitTemplate.convertSendAndReceive("X","XA", "消息来自TTL为10秒的队列:"+message);
        rabbitTemplate.convertSendAndReceive("X","XB", "消息来自TTL为40秒的队列:"+message);
    }
}

3.6消费者代码

@Slf4j
@Component
public class DeadLetterQueueConsumer {
    @RabbitListener(queues = TtlQueueConfiguration.QUEUE_D)
    public void receiveQD(Message message, Channel channel){
        //获取消息
        String msg=new String(message.getBody());
        log.info("当前时间:{},收到死信队列消息:{}",new Date(),msg);
    }
}

3.7测试

在这里插入图片描述

在这里插入图片描述

启动boot项目，在浏览器输入localhost:8080/ttl/sendMessage/Hello

在这里插入图片描述

但是这种方式有一种缺点，现在我们只有TTL为10s和40s的延迟队列，如果我们需要其他延时时间的队列的话，那么我们又得新增其他队列，这样其实并不方便，我们想要的是能够动态设置TTL，这样就不需要为每个TTL设置新的延迟队列了。

4.延迟队列优化

4.1代码结构图

在这里插入图片描述

4.2配置类

在之前写的代码基础上新增一个配置类

package com.zyh.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 MessageTtlQueueConfiguration {
    //死信交换机
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    //普通队列
    public static final String QUEUE_C = "QC";
    
    @Bean
    public Queue queueC(){
        //创建集合保存队列属性
        Map<String, Object> map = new HashMap<>();
        //设置该队列绑定的死信交换机名称
        map.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //设置routing key
        map.put("x-dead-letter-routing-key", "YD");
        //设置队列延迟时间 10秒
        map.put("x-message-ttl", 10000);
        return QueueBuilder.durable(QUEUE_C).withArguments(map).build();
    }
    
    @Bean
    public Binding queueC_BindingX(@Qualifier("queueC") Queue queue, @Qualifier("xExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("XC");
    }
}

4.3生产者

package com.zyh.controller;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
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 javax.annotation.Resource;
import javax.annotation.Resources;
import java.util.Date;

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMessageController {
    @Resource
    private RabbitTemplate rabbitTemplate;
    
    @GetMapping("/sendMessage/{message}")
    public void sendMessage(@PathVariable String message) {
        //记录日志
        log.info("当前时间:{},发送一条信息给两个TTL队列:{}", new Date(), message);
        //给QA队列发送消息
        rabbitTemplate.convertSendAndReceive("X", "XA", "消息来自TTL为10秒的队列:" + message);
        rabbitTemplate.convertSendAndReceive("X", "XB", "消息来自TTL为40秒的队列:" + message);
    }
    
    @GetMapping("/sendMessage/{message}/{ttlTime}")
    public void sendMessage(@PathVariable String message, @PathVariable String ttlTime) {
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //设置消息有效期
                message.getMessageProperties().setExpiration(ttlTime);
                return message;
            }
        };
        //记录日志
        log.info("当前时间:{},发送一条时长{}毫秒信息给队列QC:{}", new Date(),ttlTime, message);
        //给QC队列发送消息
        rabbitTemplate.convertAndSend("X", "XC", message, messagePostProcessor);
    }
}

4.4消费者

@Slf4j
@Component
public class DeadLetterQueueConsumer {
    @RabbitListener(queues = TtlQueueConfiguration.QUEUE_D)
    public void receiveQD(Message message, Channel channel){
        //获取消息
        String msg=new String(message.getBody());
        log.info("当前时间:{},收到死信队列消息:{}",new Date(),msg);
    }
}

4.5测试

启动boot项目

image.jpg

在浏览器输入

http://localhost:8080/ttl/sendMessage/Hello/20000
http://localhost:8080/ttl/sendMessage/你好/2000

image.jpg

如果在消息属性上设置TTL的方式，那么消息可能不会按时死亡，因为RabbitMQ只会检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行

到此这篇关于详解RabbitMQ延迟队列的基本使用和优化的文章就介绍到这了,更多相关RabbitMQ延迟队列内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

阅读原文内容投诉

免责声明：

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题

  • 2024年上半年信息系统项目管理师第二批次真题及答案解析（完整版）

    难度     813人已做
    查看

  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看

  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看

  • 2024年上半年软考高项第一、二批次真题考点汇总（完整版）

    难度     435人已做
    查看

  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机

详解RabbitMQ延迟队列的基本使用和优化

后端开发2023-05-19

详解RabbitMQ中死信队列和延迟队列的使用详解

后端开发2024-04-02

RabbitMQ代码篇之过期时间，死信队列，延迟队列，优先级队列的基本使用

后端开发RabbitMQ代码2024-11-30
位置：首页-资讯-后端开发
咦！没有更多了？去看看其它编程学习网 内容吧
首页课程
资料下载问答资讯