基于Rabbitmq和Redis的延迟消息实现

1 基于Rabbitmq延迟消息实现

支付时间设置为30,未支付的消息会积压在mq中,给mq带来巨大压力。我们可以利用Rabbitmq的延迟队列插件实现消息前一分钟尽快处理
在这里插入图片描述
在这里插入图片描述

1.1定义延迟消息实体

由于我们要多次发送延迟消息,因此需要先定义一个记录消息延迟时间的消息体

@Data
public class MultiDelayMessage<T> {/*** 消息体*/private T data;/*** 记录延迟时间的集合*/private List<Long> delayMillis;public MultiDelayMessage(T data, List<Long> delayMillis) {this.data = data;this.delayMillis = delayMillis;}public static <T> MultiDelayMessage<T> of(T data, Long ... delayMillis){return new MultiDelayMessage<>(data, CollUtils.newArrayList(delayMillis));}/*** 获取并移除下一个延迟时间* @return 队列中的第一个延迟时间*/public Long removeNextDelay(){return delayMillis.remove(0);}/*** 是否还有下一个延迟时间*/public boolean hasNextDelay(){return !delayMillis.isEmpty();}
}

1.2 定义常量,用于记录交换机、队列、RoutingKey等常量

package com.hmall.trade.constants;public interface MqConstants {String DELAY_EXCHANGE = "trade.delay.topic";String DELAY_ORDER_QUEUE = "trade.order.delay.queue";String DELAY_ORDER_ROUTING_KEY = "order.query";
}

1.3 抽取mq配置到nacos中

spring:rabbitmq:host: ${hm.mq.host:192.168.150.101} # 主机名port: ${hm.mq.port:5672} # 端口virtual-host: ${hm.mq.vhost:/hmall} # 虚拟主机username: ${hm.mq.un:hmall} # 用户名password: ${hm.mq.pw:123} # 密码listener:simple:prefetch: 1 # 每次只能获取一条消息,处理完成才能获取下一个消息

1.4 定义消息处理器

使用延迟消息处理器发送消息
在这里插入图片描述
在这里插入图片描述

1.5 消息监听与延迟消息再次发送

在这里插入图片描述
在这里插入图片描述

2 延迟消息实现

DelayQueue:基于JVM,保存在内存中,会出现消息丢失
在这里插入图片描述

Rabbitmq的延迟任务:基于TTL和死信交换机
在这里插入图片描述

2.1 redis的延迟任务:基于zset的去重和排序功能

在这里插入图片描述
1.为什么任务需要存储在数据库中?
延迟任务是一个通用的服务,任何有延迟需求的任务都可以调用该服务,内存数据库的存储是有限的,需要考虑数据持久化的问题,存储数据库中是一种数据安全的考虑

2.为什么使用redis中的两种数据类型,list和zset?

  • 原因一: list存储立即执行的任务,zset存储未来的数据
  • 原因二:任务量过大以后,zset的性能会下降

时间复杂度:执行时间(次数) 随着数据规模增长的变化趋势

  • 操作redis中的list命令LPUSH: 时间复杂度: O(1)
  • 操作redis中的zset命令zadd: 时间复杂度: (Mlog(n))

2.2 设计mybatis映射实体类:

/*** 版本号,用乐观锁*/@Versionprivate Integer version;乐观锁支持:
/*** mybatis-plus乐观锁支持* @return*/
@Bean
public MybatisPlusInterceptor optimisticLockerInterceptor(){MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();interceptor.addInnerInterceptor(new OptimisticLockerInnerInterceptor());return interceptor;
}

2.3 创建task类,用于接收添加任务的参数

@Data
public class Task implements Serializable {/*** 任务id*/private Long taskId;/*** 类型*/private Integer taskType;/*** 优先级*/private Integer priority;/*** 执行id*/private long executeTime;/*** task参数*/private byte[] parameters;}

2.4 添加任务

2.4.1 添加任务到数据库中

addTaskToDb(task);修改任务表和日志表

@Autowiredprivate TaskinfoMapper taskinfoMapper;@Autowiredprivate TaskinfoLogsMapper taskinfoLogsMapper;/*** 添加任务到数据库中** @param task* @return*/private boolean addTaskToDb(Task task) {boolean flag = false;try {//保存任务表Taskinfo taskinfo = new Taskinfo();BeanUtils.copyProperties(task, taskinfo);taskinfo.setExecuteTime(new Date(task.getExecuteTime()));taskinfoMapper.insert(taskinfo);//设置taskIDtask.setTaskId(taskinfo.getTaskId());//保存任务日志数据TaskinfoLogs taskinfoLogs = new TaskinfoLogs();BeanUtils.copyProperties(taskinfo, taskinfoLogs);taskinfoLogs.setVersion(1);taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);taskinfoLogsMapper.insert(taskinfoLogs);flag = true;} catch (Exception e) {e.printStackTrace();}return flag;}
2.4.2 添加任务到redis

addTaskToCache(task);判断任务执行之间是否在现在还是未来五分钟内

@Autowiredprivate CacheService cacheService;/*** 把任务添加到redis中** @param task*/private void addTaskToCache(Task task) {String key = task.getTaskType() + "_" + task.getPriority();//获取5分钟之后的时间  毫秒值Calendar calendar = Calendar.getInstance();calendar.add(Calendar.MINUTE, 5);long nextScheduleTime = calendar.getTimeInMillis();//2.1 如果任务的执行时间小于等于当前时间,存入listif (task.getExecuteTime() <= System.currentTimeMillis()) {cacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));} else if (task.getExecuteTime() <= nextScheduleTime) {//2.2 如果任务的执行时间大于当前时间 && 小于等于预设时间(未来5分钟) 存入zset中cacheService.zAdd(ScheduleConstants.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());}}

2.5 删除任务

1、删除数据库任务表,更改日志表任务状态
2、删除list或者zset中的任务

在TaskService中添加方法

/*** 取消任务* @param taskId        任务id* @return              取消结果*/
public boolean cancelTask(long taskId);
/*** 取消任务* @param taskId* @return*/
@Override
public boolean cancelTask(long taskId) {boolean flag = false;//删除任务,更新日志Task task = updateDb(taskId,ScheduleConstants.EXECUTED);//删除redis的数据if(task != null){removeTaskFromCache(task);flag = true;}return false;
}/*** 删除redis中的任务数据* @param task*/
private void removeTaskFromCache(Task task) {String key = task.getTaskType()+"_"+task.getPriority();if(task.getExecuteTime()<=System.currentTimeMillis()){cacheService.lRemove(ScheduleConstants.TOPIC+key,0,JSON.toJSONString(task));}else {cacheService.zRemove(ScheduleConstants.FUTURE+key, JSON.toJSONString(task));}
}/*** 删除任务,更新任务日志状态* @param taskId* @param status* @return*/
private Task updateDb(long taskId, int status) {Task task = null;try {//删除任务taskinfoMapper.deleteById(taskId);TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);taskinfoLogs.setStatus(status);taskinfoLogsMapper.updateById(taskinfoLogs);task = new Task();BeanUtils.copyProperties(taskinfoLogs,task);task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());}catch (Exception e){log.error("task cancel exception taskid={}",taskId);}return task;}

2.6 消费任务

1、删除list中的数据
2、使用updateDB删除任务表、跟新日志表

在TaskService中添加方法

/*** 按照类型和优先级来拉取任务* @param type* @param priority* @return*/
public Task poll(int type,int priority);

实现

/*** 按照类型和优先级拉取任务* @return*/
@Override
public Task poll(int type,int priority) {Task task = null;try {String key = type+"_"+priority;String task_json = cacheService.lRightPop(ScheduleConstants.TOPIC + key);if(StringUtils.isNotBlank(task_json)){task = JSON.parseObject(task_json, Task.class);//更新数据库信息updateDb(task.getTaskId(),ScheduleConstants.EXECUTED);}}catch (Exception e){e.printStackTrace();log.error("poll task exception");}return task;
}

2.7 未来定时任务更新-reids管道

减少与redis的交互次数
1、在引导类中添加开启任务调度注解:@EnableScheduling
2、在service中添加定时任务 @Scheduled(cron = “0 */1 * * * ?”),每分钟一次

@Scheduled(cron = "0 */1 * * * ?")
public void refresh() {System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");// 获取所有未来数据集合的key值Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_*for (String futureKey : futureKeys) { // future_250_250String topicKey = ScheduleConstants.TOPIC + futureKey.split(ScheduleConstants.FUTURE)[1];//获取该组key下当前需要消费的任务数据Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());if (!tasks.isEmpty()) {//将这些任务数据添加到消费者队列中cacheService.refreshWithPipeline(futureKey, topicKey, tasks);System.out.println("成功的将" + futureKey + "下的当前需要执行的任务数据刷新到" + topicKey + "下");}}
}
public List<Object> refreshWithPipeline(String future_key,String topic_key,Collection<String> values){List<Object> objects = stringRedisTemplate.executePipelined(new RedisCallback<Object>() {@Nullable@Overridepublic Object doInRedis(RedisConnection redisConnection) throws DataAccessException {StringRedisConnection stringRedisConnection = (StringRedisConnection)redisConnection;String[] strings = values.toArray(new String[values.size()]);stringRedisConnection.rPush(topic_key,strings);stringRedisConnection.zRem(future_key,strings);return null;}});return objects;}

总结

1、使用rebbitmq使用的场景是在支付和订单微服务中,用于实现消息可以延迟30分钟付款的功能。并借用该中间件的插件实现支付的异步下单功能,并可以快速处理前几分钟,防止消息堆积
2、使用redis是基于zset的去重和排序功能,相当于将一定数据的保存在数据库,使用定时任务同步数据库符合五分钟的任务到zset中,然后,在在zest中定时更新可以运行的任务到list集合中,相当于实现了延迟功能和缓存功能。
3、第二种还可以扩展为将rabbitmq中等待时间较长的数据存到redis中,然后定时的去同步redis中的数据到数据库中,防止消息堆积。

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

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

2.6 Windows驱动开发:使用IO与DPC定时器

本章将继续探索驱动开发中的基础部分&#xff0c;定时器在内核中同样很常用&#xff0c;在内核中定时器可以使用两种&#xff0c;即IO定时器&#xff0c;以及DPC定时器&#xff0c;一般来说IO定时器是DDK中提供的一种&#xff0c;该定时器可以为间隔为N秒做定时&#xff0c;但如…

RedCap推动5G规模应用,紫光展锐赋能产业高质量发展

5G R17 RedCap作为面向中高速物联网场景的关键技术和解决方案&#xff0c;可以大幅降低终端的复杂度、成本和功耗。在当前国内5G应用规模化发展关键时期&#xff0c;5G R17 RedCap拥有广大的市场潜力与广泛的应用场景&#xff0c;将有助于推动5G规模应用、构建融通发展的5G生态…

【海德教育】建筑八大员中综合来看,以下几个岗位比较吃香

1.施工员 施工员刚入行工资不高、劳动强度大&#xff0c;但发展前景可观。 通过考建造师等执业资格证&#xff0c;后期有望出任项目经理&#xff0c;拿到一万起步的月薪。 想要晋升项目经理、包工头等&#xff0c;对于社交能力和个人吃苦能力要求比较高&#xff0c;比较适合男孩…

7、线性数据结构-切片

切片slice 容器容量可变&#xff0c;所以长度不能定死长度可变&#xff0c;元素个数可变底层必须依赖数组&#xff0c;可以理解它依赖于顺序表&#xff0c;表现也像个可变容量和长度顺序表引用类型&#xff0c;和值类型有区别 定义切片 var s1 []int //长度、容量为0的切片&…

11月14日,每日信息差

今天是2023年11月14日&#xff0c;以下是为您准备的11条信息差 第一、中国航司首开“上海-开罗”往返直飞航线&#xff0c;12月11日首航 第二、中核集团&#xff1a;我国首次获得丰度99%以上的镱176同位素 第三、微软与谷歌将不会挑战欧盟的数据迁移法规&#xff0c;该法律将…

ProEasy机器人:运动+通讯相关说明

----------------机械手运动------- --常用指令&#xff1a;MovP弧线运动、MovL直线运动 MovP(1) --弧线轨迹运动到一号点 MovP(2) --弧线轨迹运动到二号点 MovL(1) --直线轨迹运动到一号点 MovL(2) --直线轨迹运…

LeetCode——链表(Java)

链表 简介[简单] 203. 移除链表元素[中等] 707. 设计链表[简单] 206. 反转链表[简单] 24. 两两交换链表中的节点[简单] 19. 删除链表的倒数第 N 个结点 简介 记录一下自己刷题的历程以及代码。写题过程中参考了 代码随想录。会附上一些个人的思路&#xff0c;如果有错误&#…

VB生成UTF-8的URL编码

最近在研发股票软件&#xff0c;涉及到URL字符串的处理。贴出以下代码。 Function GBtoUTF8(szInput) Dim wch, uch, szRet Dim x Dim nAsc, nAsc2, nAsc3 如果输入参数为空&#xff0c;则退出函数 If szInput "" Then GBtoUTF8 szInput Exit Function End If 开始…

前、后端程序员开发常用api接口

顺丰速运开放平台&#xff1a;可以查询顺丰速运的快递信息。京东物流开放平台&#xff1a;可以查询京东物流的快递信息。全国快递物流查询&#xff1a;1.提供包括申通、顺丰、圆通、韵达、中通、汇通等600快递公司在内的快递物流单号查询。2.与官网实时同步更新。3.自动识别快递…

【C++】join ()和detach ()函数详解和示例

简单的来说&#xff0c;join ()方法建立的线程具有阻碍作用&#xff0c;该线程不结束&#xff0c;另一些函数就无法运行。detach ()方法建立的线程&#xff0c;可以和另一些函数同时进行。下面以示例进行详细说明&#xff0c;以帮助大家理解和使用。 目录 join ()detach () jo…

PHP使用文件缓存实现html静态化

<?php // 动态生成的内容 $content "<html><body><h1>time:".date("Y-m-d H:i:s")."</h1></body></html>"; // 静态文件保存路径和文件名 $staticFilePath "file.html"; if(file_exists($s…

Ladybug 全景相机, 360°球形成像,带来全方位的视觉体验

360无死角全景照片总能给人带来强烈的视觉震撼&#xff0c;有着大片的既视感。那怎么才能拍出360球形照片呢&#xff1f;它的拍摄原理是通过图片某个点位为中心将图片其他部位螺旋式、旋转式处理&#xff0c;从而达到沉浸式体验的效果。俗话说“工欲善其事&#xff0c;必先利其…

java实现计数排序

图解 计数排序是一种线性时间复杂度的排序算法&#xff0c;它不基于比较排序&#xff0c;而是根据待排序序列中元素的值来进行排序。 具体的过程如下&#xff1a; 统计序列中每个元素出现的个数&#xff0c;得到一个计数数组count。其中&#xff0c;count[i]表示待排序序列中值…

pandas读写json的知识点

pandas对象可以直接转换为json&#xff0c;使用to_json即可。里面的orient参数很重要&#xff0c;可选值为columns,index,records,values,split,table A B C x 1 4 7 y 2 5 8 z 3 6 9 In [236]: dfjo.to_json(orient"columns") Out[236]: {"A":{"x&qu…

Centos 7rc.local脚本命令开机不执行及指定用户启动的方法

1.开机不启动 在实际生产场景中&#xff0c;我们喜欢在安装了一些软件服务后&#xff0c;将软件设置为开机自启动 配置在/etc/rc.local文件中。直接将软件服务的启动命令写在rc.local文件 注意&#xff1a;编辑完rc.local文件后&#xff0c;一定要给rc.local文件执行权限&#…

exit与return的区别 exit(1)、exit(-1)和exit(0)区别

目录 exit(); 是整个程序的结束。 exit(1)、exit(-1)和exit(0)区别 return();是某个函数的结束&#xff0c;并返回结果。 exit和_exit exit(); 是整个程序的结束。 exit&#xff08;0&#xff09;&#xff1a;正常运行程序并退出程序&#xff1b; exit&#xff08;1&am…

AIGC:使用bert_vits2实现栩栩如生的个性化语音克隆

1 VITS2模型 1.1 摘要 单阶段文本到语音模型最近被积极研究&#xff0c;其结果优于两阶段管道系统。以往的单阶段模型虽然取得了较大的进展&#xff0c;但在间歇性非自然性、计算效率、对音素转换依赖性强等方面仍有改进的空间。本文提出VITS2&#xff0c;一种单阶段的文本到…

Xilinx Kintex7中端FPGA解码MIPI视频,基于MIPI CSI-2 RX Subsystem架构实现,提供工程源码和技术支持

目录 1、前言免责声明 2、我这里已有的 MIPI 编解码方案3、本 MIPI CSI2 模块性能及其优缺点4、详细设计方案设计原理框图OV5640及其配置权电阻硬件方案MIPI CSI-2 RX SubsystemSensor Demosaic图像格式转换Gammer LUT伽马校正VDMA图像缓存AXI4-Stream toVideo OutHDMI输出 5、…

Java安全架构 JCA、JCE、JSSE、JAAS

Java语言拥有三大特征&#xff1a;平台无关性、网络移动性和安全性&#xff0c;而Java安全体系结构对这三大特征提供了强大的支持和保证&#xff0c; Java安全体系结构总共分为4个部分&#xff1a; &#xff08;1&#xff09;JCA&#xff08; Java Cryptography Architecture…

工具及方法 - 手机扫条码工具: SCANDIT APP

一般扫个链接使用微信扫一扫即可。扫具体条码&#xff0c;可以在微信里搜索小程序&#xff0c;打开也能扫&#xff0c;得到条码内容。 还有其他方式&#xff0c;比如使用淘宝、百度等APP也可以直接扫码条码&#xff0c;还能得到更多的信息。 使用百度的话&#xff0c;不扫条码…