RabbitMQ之延迟队列（手把手教你学习延迟队列）

守听  金牌会员 | 2024-7-20 23:48:32 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 904|帖子 904|积分 2712

一、延迟队列概念

延时队列,队列内部是有序的，最告急的特性就体如今它的延时属性上，延时队列中的元素是渴望在指定时间到了以后或之前取出和处理，简朴来说，延时队列就是用来存放需要在指定时间被处理的元素的队列的。
二、延迟队列使用场景



  • 1.订单在十分钟之内未支付则自动取消
  • 2.新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提示。
  • 3.用户注册乐成后，如果三天内没有登岸则进行短信提示。
  • 4.用户发起退款，如果三天内没有得到处理则关照相关运营职员。
  • 5.预定会议后，需要在预定的时间点前十分钟关照各个与会职员加入会议
这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务，如：发生订单生成事件，在十分钟之后查抄该订单支付状态，然后将未支付的订单进行关闭；看起来好像使用定时任务，不停轮询数据，每秒查一次，取出需要被处理的数据，然后处理不就完事了吗？如果数据量比力少，确实可以这样做，比如：对于“如果账单一周内未支付则进行自动结算”这样的需求，如果对于时间不是严格限定，而是宽松意义上的一周，那么每天晚上跑个定时任务查抄一下所有未支付的账单，确实也是一个可行的方案。但对于数据量比力大，并且时效性较强的场景，如：“订单十分钟内未支付则关闭“，短期内未支付的订单数据可能会有很多，运动期间甚至会到达百万甚至千万级别，对这么庞大的数据量仍然使用轮询的方式显然是不可取的，很可能在一秒内无法完成所有订单的查抄，同时会给数据库带来很大压力，无法满意业务要求而且性能低下。

三、RabbitMQ 中的 TTL

TTL 是什么呢？TTL 是 RabbitMQ 中一个消息或者队列的属性，表明一条消息或者该队列中的所有
消息的最大存活时间，单位是毫秒。换句话说，如果一条消息设置了 TTL 属性或者进入了设置 TTL 属性的队列，那么这条消息如果在 TTL 设置的时间内没有被消费，则会成为"死信"。如果同时配置了队列的 TTL 和消息的TTL，那么较小的那个值将会被使用，有两种方式设置 TTL。
1、队列设置 TTL

第一种是在创建队列的时候设置队列的“x-message-ttl”属性

2、消息设置 TTL

第二种是针对每条消息设置 TTL

3、两者的区别

如果设置了队列的 TTL 属性，那么一旦消息过期，就会被队列丢弃(如果配置了死信队列被丢到死信队列中)，而第二种方式，消息即使过期，也不一定会被马上丢弃，由于消息是否过期是在即将投递到消费者之前判断的，如果当前队列有严峻的消息积压情况，则已过期的消息大概还能存活较长时间；别的，还需要留意的一点是，如果不设置 TTL，表示消息永远不会过期，如果将 TTL 设置为 0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。
四、整合 springboot

1、添加依靠

  1.     <dependencies>
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter</artifactId>
  5.         </dependency>
  7.         <dependency>
  8.             <groupId>org.springframework.boot</groupId>
  9.             <artifactId>spring-boot-starter-test</artifactId>
  10.             <scope>test</scope>
  11.         </dependency>
  12.         <!--RabbitMQ 依赖-->
  14.         <dependency>
  15.             <groupId>org.springframework.boot</groupId>
  16.             <artifactId>spring-boot-starter-web</artifactId>
  17.         </dependency>
  19.         <dependency>
  20.             <groupId>com.alibaba</groupId>
  21.             <artifactId>fastjson</artifactId>
  22.             <version>1.2.47</version>
  23.         </dependency>
  24.         <dependency>
  25.             <groupId>org.projectlombok</groupId>
  26.             <artifactId>lombok</artifactId>
  27.         </dependency>
  28.         <!--swagger-->
  29.         <dependency>
  30.             <groupId>io.springfox</groupId>
  31.             <artifactId>springfox-swagger2</artifactId>
  32.             <version>2.9.2</version>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>io.springfox</groupId>
  36.             <artifactId>springfox-swagger-ui</artifactId>
  37.             <version>2.9.2</version>
  38.         </dependency>
  39.         <!--RabbitMQ 测试依赖-->
  40.         <dependency>
  41.             <groupId>org.springframework.amqp</groupId>
  42.             <artifactId>spring-rabbit-test</artifactId>
  43.             <scope>test</scope>
  44.         </dependency>
  45.     </dependencies>
复制代码
2、修改配置文件

  1. spring.rabbitmq.host=192.168.10.130
  2. spring.rabbitmq.port=5672
  3. spring.rabbitmq.username=guest
  4. spring.rabbitmq.password=guest
复制代码
3、添加 Swagger 配置类

  1. import org.springframework.context.annotation.Bean;
  2. import org.springframework.context.annotation.Configuration;
  3. import springfox.documentation.builders.ApiInfoBuilder;
  4. import springfox.documentation.service.ApiInfo;
  5. import springfox.documentation.service.Contact;
  6. import springfox.documentation.spi.DocumentationType;
  7. import springfox.documentation.spring.web.plugins.Docket;
  8. import springfox.documentation.swagger2.annotations.EnableSwagger2;
  9. @Configuration
  10. @EnableSwagger2
  11. public class SwaggerConfig {
  12.         @Bean
  13.         public Docket webApiConfig(){
  14.                 return new Docket(DocumentationType.SWAGGER_2)
  15.                         .groupName("webApi")
  16.                         .apiInfo(webApiInfo())
  17.                         .select()
  18.                         .build();
  19.         }
  20.         private ApiInfo webApiInfo(){
  21.                 return new ApiInfoBuilder()
  22.                         .title("rabbitmq 接口文档")
  23.                         .description("本文档描述了 rabbitmq 微服务接口定义")
  24.                         .version("1.0")
  25.                         .contact(new Contact("enjoy6288", "http://atguigu.com", "1551388580@qq.com"))
  26.                         .build();
  27.         }
  28. }
复制代码
五、队列 TTL

1、代码架构图

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

2、配置文件类代码

  1. @Configuration
  2. public class TtlQueueConfig {
  3.         public static final String X_EXCHANGE = "X";
  4.         public static final String QUEUE_A = "QA";
  5.         public static final String QUEUE_B = "QB";
  6.         public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
  7.         public static final String DEAD_LETTER_QUEUE = "QD";
  8.         // 声明 xExchange
  9.         @Bean("xExchange")
  10.         public DirectExchange xExchange(){
  11.                 return new DirectExchange(X_EXCHANGE);
  12.         }
  13.         // 声明 xExchange
  14.         @Bean("yExchange")
  15.         public DirectExchange yExchange(){
  16.                 return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
  17.         }
  18.         //声明队列 A ttl 为 10s 并绑定到对应的死信交换机
  19.         @Bean("queueA")
  20.         public Queue queueA(){
  21.                 Map<String, Object> args = new HashMap<>(3);
  22.                 //声明当前队列绑定的死信交换机
  23.                 args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
  24.                 //声明当前队列的死信路由 key
  25.                 args.put("x-dead-letter-routing-key", "YD");
  26.                 //声明队列的 TTL
  27.                 args.put("x-message-ttl", 10000);
  28.                 return QueueBuilder.durable(QUEUE_A).withArguments(args).build();
  29.         }
  30.         // 声明队列 A 绑定 X 交换机
  31.         @Bean
  32.         public Binding queueaBindingX(@Qualifier("queueA") Queue queueA,@Qualifier("xExchange") DirectExchange xExchange){
  33.                 return BindingBuilder.bind(queueA).to(xExchange).with("XA");
  34.         }
  35.         //声明队列 B ttl 为 40s 并绑定到对应的死信交换机
  36.         @Bean("queueB")
  37.         public Queue queueB(){
  38.                 Map<String, Object> args = new HashMap<>(3);
  39.                 //声明当前队列绑定的死信交换机
  40.                 args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
  41.                 //声明当前队列的死信路由 key
  42.                 args.put("x-dead-letter-routing-key", "YD");
  43.                 //声明队列的 TTL
  44.                 args.put("x-message-ttl", 40000);
  45.                 return QueueBuilder.durable(QUEUE_B).withArguments(args).build();
  46.         }
  47.         //声明队列 B 绑定 X 交换机
  48.         @Bean
  49.         public Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,@Qualifier("xExchange") DirectExchange xExchange){
  50.                 return BindingBuilder.bind(queue1B).to(xExchange).with("XB");
  51.         }
  52.         //声明死信队列 QD
  53.         @Bean("queueD")
  54.         public Queue queueD(){
  55.                 return new Queue(DEAD_LETTER_QUEUE);
  56.         }
  57.         //声明死信队列 QD 绑定关系
  58.         @Bean
  59.         public Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,@Qualifier("yExchange") DirectExchange yExchange){
  60.                 return BindingBuilder.bind(queueD).to(yExchange).with("YD");
  61.         }
  62. }
复制代码
3、消息生产者代码

  1. @Slf4j
  2. @RequestMapping("ttl")
  3. @RestController
  4. public class SendMsgController {
  5.         @Autowired
  6.         private RabbitTemplate rabbitTemplate;
  7.         @GetMapping("sendMsg/{message}")
  8.         public void sendMsg(@PathVariable String message){
  9.                 log.info("当前时间：{},发送一条信息给两个 TTL 队列:{}", new Date(), message);
  10.                 rabbitTemplate.convertAndSend("X", "XA", "消息来自 ttl 为 10S 的队列: "+message);
  11.                 rabbitTemplate.convertAndSend("X", "XB", "消息来自 ttl 为 40S 的队列: "+message);
  12.         }
  13. }
复制代码
4、消息消费者代码

  1. @Slf4j
  2. @Component
  3. public class DeadLetterQueueConsumer {
  4.         @RabbitListener(queues = "QD")
  5.         public void receiveD(Message message, Channel channel) throws IOException {
  6.                 String msg = new String(message.getBody());
  7.                 log.info("当前时间：{},收到死信队列信息{}", new Date().toString(), msg);
  8.         }
  9. }
复制代码
发起一个请求 http://localhost:8080/ttl/sendMsg/嘻嘻嘻
第一条消息在 10S 后变成了死信消息，然后被消费者消费掉，第二条消息在 40S 之后变成了死信消息，然后被消费掉，这样一个延时队列就打造完成了。
不外，如果这样使用的话，岂不是每增长一个新的时间需求，就要新增一个队列，这里只有 10S 和 40S两个时间选项，如果需要一个小时后处理，那么就需要增长 TTL 为一个小时的队列，如果是预定会议室然后提前关照这样的场景，岂不是要增长无数个队列才气满意需求？
六、延时队列优化

1、代码架构图

在这里新增了一个队列 QC,绑定关系如下,该队列不设置 TTL 时间

2、配置文件类代码

  1. @Component
  2. public class MsgTtlQueueConfig {
  3.         public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
  4.         public static final String QUEUE_C = "QC";
  5.         //声明队列 C 死信交换机
  6.         @Bean("queueC")
  7.         public Queue queueB(){
  8.                 Map<String, Object> args = new HashMap<>(3);
  9.                 //声明当前队列绑定的死信交换机
  10.                 args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
  11.                 //声明当前队列的死信路由 key
  12.                 args.put("x-dead-letter-routing-key", "YD");
  13.                 //没有声明 TTL 属性
  14.                 return QueueBuilder.durable(QUEUE_C).withArguments(args).build();
  15.         }
  16.         //声明队列 B 绑定 X 交换机
  17.         @Bean
  18.         public Binding queuecBindingX(@Qualifier("queueC") Queue queueC,@Qualifier("xExchange") DirectExchange xExchange){
  19.                 return BindingBuilder.bind(queueC).to(xExchange).with("XC");
  20.         }
  21. }
复制代码
3、消息生产者代码

  1. @GetMapping("sendExpirationMsg/{message}/{ttlTime}")
  2. public void sendMsg(@PathVariable String message,@PathVariable String ttlTime) {
  3.         rabbitTemplate.convertAndSend("X", "XC", message, correlationData ->{
  4.                 correlationData.getMessageProperties().setExpiration(ttlTime);
  5.                 return correlationData;
  6.         });
  7.         log.info("当前时间：{},发送一条时长{}毫秒 TTL 信息给队列 C:{}", new Date(),ttlTime, message);
  8. }
复制代码
发起请求


  • http://localhost:8080/ttl/sendExpirationMsg/你好 1/20000
  • http://localhost:8080/ttl/sendExpirationMsg/你好 2/2000

看起来好像没什么问题，但是在最开始的时候，就介绍过如果使用在消息属性上设置 TTL 的方式，消息可能并不会按时“死亡“，由于 RabbitMQ 只会查抄第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行。
七、Rabbitmq 插件实现延迟队列

1、安装延时队列插件

由于博主是用Docker安装的RabbitMQ，所以安装延时队列插件也是在Docker中进行。
首先在官网下载rabbitmq_delayed_message_exchange 插件。
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

点击下载.ez文件。
然后通过自己的文件传输工具将.ez文件上传到假造机中，博主这里将.ez文件放到了/mnt目录下。

然后输入sudo docker ps命令查看自己的rabbitmq是否正在运行，如果不在运行则输入sudo docker start idid这里填你自己的容器id，如果不知道自己id的，输入sudo docker pa -a查看。

当容器运行起来后，输入sudo docker cp /mnt/rabbitmq_delayed_message_exchange-3.12.0.ez rabbit:/plugins命令，将刚插件拷贝到容器内plugins目录下。

拷贝完成后，输入sudo docker exec -it rabbit /bin/bash命令，进入容器。

在容器内plugins目录下，查看插件是否上传乐成ls -l|grep delay

然后启动插件，在当前目录下输入rabbitmq-plugins enable rabbitmq_delayed_message_exchange命令

到这里插件安装就完成了，接下来我们需要重启RabbitMQ容器。执行exit命令退出RabbitMQ容器内部，然后执行docker restart rabbit命令重启RabbitMQ容器

在容器重启完成后，我们可以登录RabbitMQ的Web端管理界面，在Exchanges选项卡下，点击Add a new exchange，在Type内里看是否出现了x-delayed-message选项，如下图到这里，整个安装过程就完毕了。

2、代码架构图

在这里新增了一个队列 delayed.queue,一个自界说互换机 delayed.exchange，绑定关系如下:

3、配置文件类代码

在我们自界说的互换机中，这是一种新的互换类型，该类型消息支持延迟投递机制 消息传递后并不会立刻投递到目标队列中，而是存储在 mnesia(一个分布式数据系统)表中，当到达投递时间时，才投递到目标队列中。
  1. @Configuration
  2. public class DelayedQueueConfig {
  3.         public static final String DELAYED_QUEUE_NAME = "delayed.queue";
  4.         public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
  5.         public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
  6.         @Bean
  7.         public Queue delayedQueue() {
  8.                 return new Queue(DELAYED_QUEUE_NAME);
  9.         }
  10.         //自定义交换机 我们在这里定义的是一个延迟交换机
  11.         @Bean
  12.          public CustomExchange delayedExchange() {
  13.                 Map<String, Object> args = new HashMap<>();
  14.                 //自定义交换机的类型
  15.                 args.put("x-delayed-type", "direct");
  16.                 return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, args);
  17.         }
  18.         @Bean
  19.         public Binding bindingDelayedQueue(@Qualifier("delayedQueue") Queue queue,@Qualifier("delayedExchange") CustomExchange delayedExchange) {
  20.                 return BindingBuilder.bind(queue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
  21.         }
  22. }
复制代码
4、消息生产者代码

  1. public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
  2. public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
  3. @GetMapping("sendDelayMsg/{message}/{delayTime}")
  4. public void sendMsg(@PathVariable String message,@PathVariable Integer delayTime) {
  5.         rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, message, correlationData ->{
  6.                 correlationData.getMessageProperties().setDelay(delayTime);
  7.                 return correlationData;
  8.         });
  9.         log.info(" 当 前 时 间 ： {}, 发送一条延迟 {} 毫秒的信息给队列 delayed.queue:{}", new
  10. Date(),delayTime, message);
  11. }
复制代码
5、消息消费者代码

  1. public static final String DELAYED_QUEUE_NAME = "delayed.queue";
  2. @RabbitListener(queues = DELAYED_QUEUE_NAME)
  3. public void receiveDelayedQueue(Message message){
  4.         String msg = new String(message.getBody());
  5.         log.info("当前时间：{},收到延时队列的消息：{}", new Date().toString(), msg);
  6. }
复制代码
发起请求：


  • http://localhost:8080/ttl/sendDelayMsg/come on baby1/20000
  • http://localhost:8080/ttl/sendDelayMsg/come on baby2/2000

第二个消息被先消费掉了，符合预期
总结

延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的使用RabbitMQ 的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被精确处理的消息不会被丢弃。别的，通过 RabbitMQ 集群的特性，可以很好的解决单点故障问题，不会由于单个节点挂掉导致延时队列不可用或者消息丢失。
当然，延时队列还有很多其它选择，比如使用 Java 的 DelayQueue，使用 Redis 的 zset，使用 Quartz或者使用 kafka 的时间轮，这些方式各有特点,看需要实用的场景

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

守听

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表