搜索

延迟队列实现，定时任务，关闭订单

Java技术驿站

共 6230字，需浏览 13分钟

 · 2020-10-02

场景

开发中经常需要用到定时任务，对于商城来说，定时任务尤其多，比如优惠券定时过期、订单定时关闭、微信支付2小时未支付关闭订单等等，都需要用到定时任务，但是定时任务本身有一个问题，一般来说我们都是通过定时轮询查询数据库来判断是否有任务需要执行，也就是说不管怎么样，我们需要先查询数据库，而且有些任务对时间准确要求比较高的，需要每秒查询一次，对于系统小倒是无所谓，如果系统本身就大而且数据也多的情况下，这就不大现实了，所以需要其他方式的，当然实现的方式有多种多样的，比如Redis实现定时队列、基于优先级队列的JDK延迟队列、时间轮等。因为我们项目中本身就使用到了Rabbitmq，所以基于方便开发和维护的原则，我们使用了Rabbitmq延迟队列来实现定时任务,不知道rabbitmq是什么的和不知道springboot怎么集成Rabbitmq的可以查看我之前的文章Spring boot集成RabbitMQ

Rabbitmq延迟队列

Rabbitmq本身是没有延迟队列的，只能通过Rabbitmq本身队列的特性来实现，想要Rabbitmq实现延迟队列，需要使用Rabbitmq的死信交换机（Exchange）和消息的存活时间TTL（Time To Live）

死信交换机

一个消息在满足如下条件下，会进死信交换机，记住这里是交换机而不是队列，一个交换机可以对应很多队列。

一个消息被Consumer拒收了，并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里，被其他消费者使用。

上面的消息的TTL到了，消息过期了。

队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上。

死信交换机就是普通的交换机，只是因为我们把过期的消息扔进去，所以叫死信交换机，并不是说死信交换机是某种特定的交换机

消息TTL（消息存活时间）

消息的TTL就是消息的存活时间。RabbitMQ可以对队列和消息分别设置TTL。对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。如果队列设置了，消息也设置了，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息死亡的时间有可能不一样（不同的队列设置）。这里单讲单个消息的TTL，因为它才是实现延迟任务的关键。

  1. byte[] messageBodyBytes = "Hello, world!".getBytes();  
  2. AMQP.BasicProperties properties = new AMQP.BasicProperties();  
  3. properties.setExpiration("60000");  
  4. channel.basicPublish("my-exchange", "queue-key", properties, messageBodyBytes);  

可以通过设置消息的expiration字段或者x-message-ttl属性来设置时间，两者是一样的效果。只是expiration字段是字符串参数，所以要写个int类型的字符串：当上面的消息扔到队列中后，过了60秒，如果没有被消费，它就死了。不会被消费者消费到。这个消息后面的，没有“死掉”的消息对顶上来，被消费者消费。死信在队列中并不会被删除和释放，它会被统计到队列的消息数中去

处理流程图

创建交换机（Exchanges）和队列（Queues）

创建死信交换机

如图所示，就是创建一个普通的交换机，这里为了方便区分，把交换机的名字取为：delay

创建自动过期消息队列

这个队列的主要作用是让消息定时过期的，比如我们需要2小时候关闭订单，我们就需要把消息放进这个队列里面，把消息过期时间设置为2小时创建一个一个名为delay_queue1的自动过期的队列，当然图片上面的参数并不会让消息自动过期，因为我们并没有设置x-message-ttl参数，如果整个队列的消息有消息都是相同的，可以设置，这里为了灵活，所以并没有设置，另外两个参数x-dead-letter-exchange代表消息过期后，消息要进入的交换机，这里配置的是delay，也就是死信交换机，x-dead-letter-routing-key是配置消息过期后，进入死信交换机的routing-key,跟发送消息的routing-key一个道理，根据这个key将消息放入不同的队列

创建消息处理队列

这个队列才是真正处理消息的队列，所有进入这个队列的消息都会被处理消息队列的名字为delay_queue2

消息队列绑定到交换机

进入交换机详情页面，将创建的2个队列（delayqueue1和delayqueue2）绑定到交换机上面自动过期消息队列的routing key 设置为delay

绑定delayqueue2delayqueue2 的key要设置为创建自动过期的队列的x-dead-letter-routing-key参数，这样当消息过期的时候就可以自动把消息放入delay_queue2这个队列中了

绑定后的管理页面如下图：

当然这个绑定也可以使用代码来实现，只是为了直观表现，所以本文使用的管理平台来操作

发送消息

  1. String msg = "hello word";  
  2. MessageProperties messageProperties = newMessageProperties();  
  3.         messageProperties.setExpiration("6000");
  4.         messageProperties.setCorrelationId(UUID.randomUUID().toString().getBytes());
  5.         Message message = newMessage(msg.getBytes(), messageProperties);
  6.         rabbitTemplate.convertAndSend("delay", "delay",message);

主要的代码就是

  1. messageProperties.setExpiration("6000");  

设置了让消息6秒后过期

注意：因为要让消息自动过期，所以一定不能设置delay_queue1的监听，不能让这个队列里面的消息被接受到，否则消息一旦被消费，就不存在过期了

接收消息

接收消息配置好delay_queue2的监听就好了

  1. package wang.raye.rabbitmq.demo1;
  2. import org.springframework.amqp.core.AcknowledgeMode;  
  3. import org.springframework.amqp.core.Binding;  
  4. import org.springframework.amqp.core.BindingBuilder;  
  5. import org.springframework.amqp.core.DirectExchange;  
  6. import org.springframework.amqp.core.Message;  
  7. import org.springframework.amqp.core.Queue;  
  8. import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;  
  9. import org.springframework.amqp.rabbit.connection.ConnectionFactory;  
  10. import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;  
  11. import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;  
  12. import org.springframework.beans.factory.annotation.Autowired;  
  13. import org.springframework.context.annotation.Bean;  
  14. import org.springframework.context.annotation.Configuration;
  15. @Configuration
  16. publicclassDelayQueue{  
  17.     /** 消息交换机的名字*/
  18.     publicstaticfinalString EXCHANGE = "delay";
  19.     /** 队列key1*/
  20.     publicstaticfinalString ROUTINGKEY1 = "delay";
  21.     /** 队列key2*/
  22.     publicstaticfinalString ROUTINGKEY2 = "delay_key";
  23.     /**
  24.      * 配置链接信息
  25.      * @return
  26.      */
  27.     @Bean
  28.     publicConnectionFactory connectionFactory() {
  29.         CachingConnectionFactory connectionFactory = newCachingConnectionFactory("120.76.237.8",5672);
  30.         connectionFactory.setUsername("kberp");
  31.         connectionFactory.setPassword("kberp");
  32.         connectionFactory.setVirtualHost("/");
  33.         connectionFactory.setPublisherConfirms(true); // 必须要设置
  34.         return connectionFactory;
  35.     }
  36.     /**  
  37.      * 配置消息交换机
  38.      * 针对消费者配置  
  39.         FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念  
  40.         HeadersExchange ：通过添加属性key-value匹配  
  41.         DirectExchange:按照routingkey分发到指定队列  
  42.         TopicExchange:多关键字匹配  
  43.      */  
  44.     @Bean  
  45.     publicDirectExchange defaultExchange() {  
  46.         returnnewDirectExchange(EXCHANGE, true, false);
  47.     }
  48.     /**
  49.      * 配置消息队列2
  50.      * 针对消费者配置  
  51.      * @return
  52.      */
  53.     @Bean
  54.     publicQueue queue() {  
  55.        returnnewQueue("delay_queue2", true); //队列持久  
  56.     }
  57.     /**
  58.      * 将消息队列2与交换机绑定
  59.      * 针对消费者配置  
  60.      * @return
  61.      */
  62.     @Bean  
  63.     @Autowired
  64.     publicBinding binding() {  
  65.         returnBindingBuilder.bind(queue()).to(defaultExchange()).with(DelayQueue.ROUTINGKEY2);  
  66.     }
  67.     /**
  68.      * 接受消息的监听，这个监听会接受消息队列1的消息
  69.      * 针对消费者配置  
  70.      * @return
  71.      */
  72.     @Bean  
  73.     @Autowired
  74.     publicSimpleMessageListenerContainer messageContainer2(ConnectionFactory connectionFactory) {  
  75.         SimpleMessageListenerContainer container = newSimpleMessageListenerContainer(connectionFactory());  
  76.         container.setQueues(queue());  
  77.         container.setExposeListenerChannel(true);  
  78.         container.setMaxConcurrentConsumers(1);  
  79.         container.setConcurrentConsumers(1);  
  80.         container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认  
  81.         container.setMessageListener(newChannelAwareMessageListener() {
  82.             publicvoid onMessage(Message message, com.rabbitmq.client.Channel channel) throwsException{
  83.                 byte[] body = message.getBody();  
  84.                 System.out.println("delay_queue2 收到消息 : "+ newString(body));  
  85.                 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费  
  86.             }  
  87.         });  
  88.         return container;  
  89.     }  
  90. }

在消息监听中处理需要定时处理的任务就好了，因为Rabbitmq能发送消息，所以可以把任务特征码发过来，比如关闭订单就把订单id发过来，这样就避免了需要查询一下那些订单需要关闭而加重MySQL负担了，毕竟一旦订单量大的话，查询本身也是一件很费IO的事情

总结

基于Rabbitmq实现定时任务，就是将消息设置一个过期时间，放入一个没有读取的队列中，让消息过期后自动转入另外一个队列中，监控这个队列消息的监听处来处理定时任务具体的操作

浏览 23
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
15种时间序列预测方法总结(包含多种方法代码实现)
向AI转型的程序员都关注了这个号👇👇👇在这篇文章中，我们将深入探讨时间序列预测的基本概念和方法。我们将首先介绍单元预测和多元预测的概念，然后详细介绍各种深度学习和传统机器学习方法如何应用于时间序列预测，包括循环神经网络（RNN）、一维卷积神经网络（1D-CNN）、Transformer、自回归模型（
机器学习AI算法工程
0
SpringBoot 实现图片防盗链功能
程序员的成长之路互联网/程序员/技术/资料共享 关注阅读本文大概需要 4 分钟。来自：blog.csdn.net/weixin_46157208/article/details/138051737前言出于安全考虑，我们需要后端返回的图片只允许在某个网站内展示，不想被爬虫拿到图片地
程序员的成长之路
0
一站式解决方案：基于 Arthas 实现服务发现和权限控制
来源：juejin.cn/post/7281849496983994383👉 欢迎加入小哈的星球 ，你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 /  赠书福利全栈前后端分离博客项目 2.0 版本完结啦， 演示链接
小哈学Java
0
用 Shader 实现旗帜飘扬动画效果
我觉得对于刚入门 3D 编程的朋友来说，如果能够完成代码创建模型数据->创建材质->编写Shader动画这一系列，想必会有满满的成就感。今天就用 Cocos Creator 的 utils.MeshUtils.createMesh 接口，带大家感受一下这个流程。这个流程不仅可以用于新手学
COCOS
2
互联网晚报 | ​央视发文谈调休：原则是最小干预；特斯拉中国全系降价；二手交易平台现大量转卖SU7创始车型订单
央视发文谈调休：原则是最小干预“五一其实只放一天”近日冲上热搜，再度引发“假期该不该调休”的热议。如何调休，也有讲究。一个基本原则是，“最小干预，尽量不打乱人们的正常生活节奏”。我国现行法定节假日总天数为11天，周末休息104天，带薪休假5～15天，总假期天数在国际上处于中等偏上水平。专家指出，“之
产品刘
0
SpringBoot+Minio实现上传凭证、分片上传、秒传和断点续传
关注我们,设为星标,每天7:40不见不散,架构路上与您共享回复架构师获取资源大家好，我是你们的朋友架构君，一个会写代码吟诗的架构师。Spring Boot整合Minio后，前端的文件上传有两种方式：1、文件上传到后端，由后端保存到Minio这种方式好处是完全由后端集中管理，可以很好的做到、身份验证、
Java架构师社区
0
超越原生，散点图实现华夫饼图
之前我们介绍过了如何使用新卡片图实现华夫饼图。参考：超越原生，PowerBI 华夫饼图实现但是利用卡片图实现的华夫饼图有一些缺点，形状之间的大小跟间距不太好把握，而且有时形状大一点的话显示就会不正常，需要做出二次调整。今天给大家介绍一种原生视觉对象生成华夫饼图的更佳方案，既简单又美观。上图是利用散点
PowerBI战友联盟
2
Spring Boot + flowable 快速实现工作流
关注我们,设为星标,每天7:40不见不散,架构路上与您共享回复架构师获取资源大家好，我是你们的朋友架构君，一个会写代码吟诗的架构师。来源：blog.csdn.net/zhan107876/article/details/120815560总览一、flowable-ui部署运行二、绘制流程图绘图细节：
Java架构师社区
0
实现订单30分钟自动取消的策略
原文：juejin.cn/post/7285167401821798400简介在电商和其他涉及到在线支付的应用中，通常需要实现一个功能：如果用户在生成订单后的一定时间内未完成支付，系统将自动取消该订单。本文将详细介绍基于Spring Boot框架实现订单30分钟内未支付自动取消的几种方案，并提供实例
JAVA乐园
0
AI大模型之路 第三篇：从零实现词嵌入模型，加深理解！
你好，我是郭震今天我们研究「AI大模型第三篇」：词维度预测，很多读者听过词嵌入，这篇文章解答下面问题：词嵌入是什么意思？怎么做到的？原理是什么？从零实现一个专属你数据集的词嵌入我们完整从零走一遍，根基的东西要理解透，这样才能发明出更好的东西。1 skip-gram模型Skip-gram模型是一种广泛
Python与算法社区
11
点赞
评论
收藏
分享

手机扫一扫分享

举报