javaredis延迟队列_Redis实现延迟队列

更新时间:2023-07-28 10:12:12 阅读: 评论:0

javaredis延迟队列_Redis实现延迟队列
延迟队列,顾名思义它是⼀种带有延迟功能的消息队列。那么,是在什么场景下我才需要这样的队列呢?
1. 背景
我们先看看以下业务场景:
当订单⼀直处于未⽀付状态时,如何及时的关闭订单
如何定期检查处于退款状态的订单是否已经退款成功
在订单长时间没有收到下游系统的状态通知的时候,如何实现阶梯式的同步订单状态的策略
在系统通知上游系统⽀付成功终态时,上游系统返回通知失败,如何进⾏异步通知实⾏分频率发送:15s 3m 10m 30m 30m 1h 2h 6h
15h
1.1 解决⽅案
最简单的⽅式,定时扫表。例如对于订单⽀付失效要求⽐较⾼的,每2S扫表⼀次检查过期的订单进⾏主动关单操作。优点是简单,缺点是每分钟全局扫表,浪费资源,如果遇到表数据订单量即将过期的订单量很⼤,会造成关单延迟。
使⽤RabbitMq或者其他MQ改造实现延迟队列,优点是,开源,现成的稳定的实现⽅案,缺点是:MQ是⼀个消息中间件,如果团队技术栈本来就有MQ,那还好,如果不是,那为了延迟队列⽽去部署⼀套MQ成本有点⼤
使⽤Redis的zt、list的特性,我们可以利⽤redis来实现⼀个延迟队列RedisDelayQueue
2. 设计⽬标
实时性:允许存在⼀定时间的秒级误差
⾼可⽤性:⽀持单机、⽀持集群
⽀持消息删除:业务会随时删除指定消息
消息可靠性:保证⾄少被消费⼀次
消息持久化:基于Redis⾃⾝的持久化特性,如果Redis数据丢失,意味着延迟消息的丢失,不过可以做主备和集群保证。这个可以考虑后续优化将消息持久化到MangoDB中
3. 设计⽅案
设计主要包含以下⼏点:
将整个Redis当做消息池,以KV形式存储消息
使⽤ZSET做优先队列,按照Score维持优先级
使⽤LIST结构,以先进先出的⽅式消费打开时空之门
ZSET和LIST存储消息地址(对应消息池的每个KEY)
⾃定义路由对象,存储ZSET和LIST名称,以点对点的⽅式将消息从ZSET路由到正确的LIST
使⽤定时器维护路由
根据TTL规则实现消息延迟
3.1 设计图
还是基于有赞的延迟队列设计,进⾏优化改造及代码实现。有赞设计
图⽚仅供参考,基本可以描述整个流程的执⾏过程,图⽚源于⽂末的参考博客中
3.3 任务的⽣命周期
新增⼀个JOB,会在ZING:DELAY_QUEUE:JOB_POOL中插⼊⼀条数据,记录了业务⽅消费⽅。ZING:DELAY_QUEUE:BUCKET也会插⼊⼀条记录,记录执⾏的时间戳性格分析
搬运线程会去ZING:DELAY_QUEUE:BUCKET中查找哪些执⾏时间戳的RunTimeMillis⽐现在的时间⼩,将这些记录全部删除;同时会解析出每个任务的Topic是什么,然后将这些任务PUSH到TOPIC对应的列表ZING:DELAY_QUEUE:QUEUE中
每个TOPIC的LIST都会有⼀个监听线程去批量获取LIST中的待消费数据,获取到的数据全部扔给这个TOPIC的消费线程池
消费线程池执⾏会去ZING:DELAY_QUEUE:JOB_POOL查找数据结构,返回给回调结构,执⾏回调⽅法。
3.4 设计要点
3.4.1 基本概念
JOB:需要异步处理的任务,是延迟队列⾥的基本单元
Topic:⼀组相同类型Job的集合(队列)。供消费者来订阅
3.4.2 消息结构
每个JOB必须包含以下⼏个属性
jobId:Job的唯⼀标识。⽤来检索和删除指定的Job信息
topic:Job类型。可以理解成具体的业务名称
delay:Job需要延迟的时间。单位:秒。(服务端会将其转换为绝对时间)
body:Job的内容,供消费者做具体的业务处理,以json格式存储
retry:失败重试次数
url:通知URL
3.5 设计细节
3.5.1 如何快速消费ZING:DELAY_QUEUE:QUEUE
面瘫按摩手法图
最简单的实现⽅式就是使⽤定时器进⾏秒级扫描,为了保证消息执⾏的时效性,可以设置每1S请求Redis⼀次,判断队列中是否有待消费的JOB。但是这样会存在⼀个问题,如果queue中⼀直没有可消费的JOB,那频繁的扫描就失去了意义,也浪费了资源,幸好LIST中有⼀个BLPOP阻塞原语,如果list中有数据就会⽴马返回,如果没有数据就会⼀直阻塞在那⾥,直到有数据返回,可以设置阻塞的超时时间,超时会返回NULL;具体的实现⽅式及策略会在代码中进⾏具体的实现介绍
3.5.2 避免定时导致的消息重复搬运及消费
使⽤Redis的分布式锁来控制消息的搬运,从⽽避免消息被重复搬运导致的问题
使⽤分布式锁来保证定时器的执⾏频率
4. 核⼼代码实现
4.1 技术说明
技术栈:SpringBoot,Redisson,Redis,分布式锁,定时器
注意:本项⽬没有实现设计⽅案中的多Queue消费,只开启了⼀个QUEUE,这个待以后优化
4.2 核⼼实体
4.2.1 Job新增对象
/**
* 消息结构
*
* @author 睁眼看世界
* @date 2020年1⽉15⽇
*/
@Data
public class Job implements Serializable {
private static final long rialVersionUID = 1L;
柳条怎么画/**
* Job的唯⼀标识。⽤来检索和删除指定的Job信息
*/
@NotBlank
private String jobId;
/**
* Job类型。可以理解成具体的业务名称
*/
@NotBlank
private String topic;
/
**
* Job需要延迟的时间。单位:秒。(服务端会将其转换为绝对时间) */
private Long delay;
/**
* Job的内容,供消费者做具体的业务处理,以json格式存储
*/
@NotBlank
private String body;
/**
* 失败重试次数
*/
private int retry = 0;
/**
* 通知URL
*/
@NotBlank
private String url;
}
4.2.2 Job删除对象
/**
* 消息结构
*
* @author 睁眼看世界
* @date 2020年1⽉15⽇
*/
其不善者而改之@Data
public class JobDie implements Serializable { private static final long rialVersionUID = 1L; /**
* Job的唯⼀标识。⽤来检索和删除指定的Job信息*/
@NotBlank
private String jobId;
关于春节的一幅画/**
* Job类型。可以理解成具体的业务名称
*/
@NotBlank4年级
private String topic;
}
4.3 搬运线程
/**
* 搬运线程
*
* @author 睁眼看世界
* @date 2020年1⽉17⽇
*/
@Slf4j紫水晶的功效
@Component
public class CarryJobScheduled {
@Autowired
private RedissonClient redissonClient;
/**
* 启动定时开启搬运JOB信息
*/
@Scheduled(cron = "*/1 * * * * *")
public void carryJobToQueue() {

本文发布于:2023-07-28 10:12:12,感谢您对本站的认可!

本文链接:https://www.wtabcd.cn/fanwen/fan/89/1099824.html

版权声明:本站内容均来自互联网,仅供演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,我们将在24小时内删除。

标签:消息   延迟   队列   实现   订单   业务   时间
相关文章
留言与评论(共有 0 条评论)
   
验证码:
推荐文章
排行榜
Copyright ©2019-2022 Comsenz Inc.Powered by © 专利检索| 网站地图