java + redis zset实现延迟队列(定时到期执行任务)
lipiwang 2024-11-22 17:20 5 浏览 0 评论
在Redis中,zet作为有序集合,可以利用其有序的特性,将任务添加到zset中,将任务的到期时间作为score,利用zset的默认有序特性,zrangewithscores可以获取score值最小的元素(也就是最近到期的任务),判断系统时间与该任务的到期时间大小,如果达到到期时间,就执行业务,并删除该到期任务,继续判断下一个元素,如果没有到期,就sleep一段时间(比如1秒),如果集合为空,也sleep一段时间。
1. 添加依赖
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.3.0</version>
</dependency>
2. 测试代码
package com.demo;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Tuple;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;
/**
* 基于redis的延迟队列
*/
public class RedisDelayQueue {
public static void main(String[] args) {
System.out.println("begin time:" + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
RedisProduceThread produceThread=new RedisProduceThread();
produceThread.start();
RedisConsumeThread consumeThread=new RedisConsumeThread();
consumeThread.start();
}
public static class DelayTask {
/* 触发时间*/
private long time;
private String name;
public long getTime() {
return time;
}
public void setTime(long time) {
this.time = time;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
// 添加任务线程
public static class RedisProduceThread extends Thread {
public RedisProduceThread() {
}
@Override
public void run() {
Jedis jedis = new Jedis("127.0.0.1",6379);
while (true)
{
long timeMillis = System.currentTimeMillis();
Random rnd = new Random();
int i = rnd.nextInt(30);
double delay = timeMillis / 1000 + i;
jedis.zadd("myzset", delay, "item-" + i);
Double doubleDelay = delay;
long longDelay = doubleDelay.longValue();
System.out.println("添加业务:item-" + i + ",添加时间:" + timeMillis / 1000 + " ,到期时间:" + longDelay + ",延迟时间:" + i + " 秒");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
// 读取到期任务线程
public static class RedisConsumeThread extends Thread {
public RedisConsumeThread() {
}
@Override
public void run() {
Jedis jedis = new Jedis("127.0.0.1",6379);
while (true) {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
// 从redis读取时间最小的数据
long timestamp = System.currentTimeMillis() / 1000;
Set<Tuple> myzset = jedis.zrangeWithScores("myzset", 0, 1);
// 如果读取记录为空
if(myzset.isEmpty())
{
// 延时1秒
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
continue;
}
Iterator<Tuple> iterator = myzset.iterator();
while (iterator.hasNext())
{
Tuple tuple = iterator.next();
String item = tuple.getElement();
Double score = tuple.getScore();
// 如果当前记录到期
if(timestamp >= score)
{
long lscore = score.longValue();
// 执行业务处理
System.out.println("到期业务:" + item + " ,到期时间:" + lscore + ",系统时间:" + timestamp);
// 处理完成后,删除当前记录
jedis.zrem("myzset", item);
// 继续循环读取下一条
}
else
{
// 最小记录未到期,延时1秒
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
}
}
3. 执行测试
添加业务:item-1,添加时间:1645515070 ,到期时间:1645515071,延迟时间:1 秒
到期业务:item-5 ,到期时间:1645515069,系统时间:1645515070
到期业务:item-1 ,到期时间:1645515071,系统时间:1645515071
添加业务:item-5,添加时间:1645515073 ,到期时间:1645515078,延迟时间:5 秒
到期业务:item-15 ,到期时间:1645515073,系统时间:1645515074
添加业务:item-23,添加时间:1645515076 ,到期时间:1645515099,延迟时间:23 秒
添加业务:item-11,添加时间:1645515079 ,到期时间:1645515090,延迟时间:11 秒
到期业务:item-5 ,到期时间:1645515078,系统时间:1645515079
添加业务:item-5,添加时间:1645515082 ,到期时间:1645515087,延迟时间:5 秒
添加业务:item-7,添加时间:1645515085 ,到期时间:1645515092,延迟时间:7 秒
添加业务:item-29,添加时间:1645515088 ,到期时间:1645515117,延迟时间:29 秒
到期业务:item-20 ,到期时间:1645515087,系统时间:1645515088
到期业务:item-5 ,到期时间:1645515087,系统时间:1645515088
到期业务:item-11 ,到期时间:1645515090,系统时间:1645515090
可以看到添加业务的时间加上延迟时间就是业务到期时间,在业务到期的下一秒,就输出了到期提示。
可以根据业务量的大小,每次读取的数据可以是一条数据,也可以是多条数据。一般情况下,每秒做一次检查可以满足大多数的业务需要,特殊情况下,可以将sleep的时间缩小(比如500ms或者300ms),这样可以做到更大的精确性。
- 上一篇:Java基础——泛型
- 下一篇:Flink DataStream API编程指南
相关推荐
- ubuntu单机安装open-falcon极度详细操作
-
备注:以下操作均由本人实际操作并得到验证,喜欢的同学可尝试操作安装。步骤一1.1环境准备(使用系统:ubuntu18.04)1.1.1安装redisubuntu下安装(参考借鉴:https://...
- Linux搭建promtail、loki、grafana轻量日志监控系统
-
一:简介日志监控告警系统,较为主流的是ELK(Elasticsearch、Logstash和Kibana核心套件构成),虽然优点是功能丰富,允许复杂的操作。但是,这些方案往往规模复杂,资源占用高,...
- 一文搞懂,WAF阻止恶意攻击的8种方法
-
WAF(Web应用程序防火墙)是应用程序和互联网流量之间的第一道防线,它监视和过滤Internet流量以阻止不良流量和恶意请求,WAF是确保Web服务的可用性和完整性的重要安全解决方案。它...
- 14配置appvolume(ios14.6配置文件)
-
使用AppVolumes应用程序功能,您可以管理应用程序的整个生命周期,包括打包、更新和停用应用程序。您还可以自定义应用程序分配,以向最终用户提供应用程序的特定版本14.1安装appvolume...
- 目前流行的缺陷管理工具(缺陷管理方式存在的优缺点)
-
摘自:https://blog.csdn.net/jasonteststudy/article/details/7090127?utm_medium=distribute.pc_relevant.no...
- 开源数字货币交易所开发学习笔记(2)——SpringCloud
-
前言码云(Gitee)上开源数字货币交易所源码CoinExchange的整体架构用了SpringCloud,对于经验丰富的Java程序员来说,可能很简单,但是对于我这种入门级程序员,还是有学习的必要的...
- 开发JAX-RPC Web Services for WebSphere(下)
-
在开发JAX-RPCWebServicesforWebSphere(上)一文中,小编为大家介绍了如何创建一个Web服务项目、如何创建一个服务类和Web服务,以及部署项目等内容。接下来小编将为大...
- CXF学习笔记1(cxf client)
-
webservice是发布服务的简单并实用的一种技术了,个人学习了CXF这个框架,也比较简单,发布了一些笔记,希望对笔友收藏并有些作用哦1.什么是webServicewebService让一个程序可...
- 分布式RPC最全详解(图文全面总结)
-
分布式通信RPC是非常重要的分布式系统组件,大厂经常考察的Dubbo等RPC框架,下面我就全面来详解分布式通信RPC@mikechen本篇已收于mikechen原创超30万字《阿里架构师进阶专题合集》...
- Oracle WebLogic远程命令执行0day漏洞(CVE-2019-2725补丁绕过)预警
-
概述近日,奇安信天眼与安服团队通过数据监控发现,野外出现OracleWebLogic远程命令执行漏洞最新利用代码,此攻击利用绕过了厂商今年4月底所发布的最新安全补丁(CVE-2019-2725)。由...
- Spring IoC Container 原理解析(spring中ioc三种实现原理)
-
IoC、DI基础概念关于IoC和DI大家都不陌生,我们直接上martinfowler的原文,里面已经有DI的例子和spring的使用示例《InversionofControlContainer...
- Arthas线上服务器问题排查(arthas部署)
-
1Arthas(阿尔萨斯)能为你做什么?这个类从哪个jar包加载的?为什么会报各种类相关的Exception?我改的代码为什么没有执行到?难道是我没commit?分支搞错了?遇到问题无法在...
- 工具篇之IDEA功能插件HTTP_CLENT(idea2021插件)
-
工具描述:Java开发人员通用的开发者工具IDEA集成了HTTPClient功能,之后可以无需单独安装使用PostMan用来模拟http请求。创建方式:1)简易模式Tools->HTTPCl...
- RPC、Web Service等几种远程监控通信方式对比
-
几种远程监控通信方式的介绍一.RPCRPC使用C/S方式,采用http协议,发送请求到服务器,等待服务器返回结果。这个请求包括一个参数集和一个文本集,通常形成“classname.meth...
- 《github精选系列》——SpringBoot 全家桶
-
1简单总结1SpringBoot全家桶简介2项目简介3子项目列表4环境5运行6后续计划7问题反馈gitee地址:https://gitee.com/yidao620/springbo...
你 发表评论:
欢迎- 一周热门
- 最近发表
- 标签列表
-
- maven镜像 (69)
- undefined reference to (60)
- zip格式 (63)
- oracle over (62)
- date_format函数用法 (67)
- 在线代理服务器 (60)
- shell 字符串比较 (74)
- x509证书 (61)
- localhost (65)
- java.awt.headless (66)
- syn_sent (64)
- settings.xml (59)
- 弹出窗口 (56)
- applicationcontextaware (72)
- my.cnf (73)
- httpsession (62)
- pkcs7 (62)
- session cookie (63)
- java 生成uuid (58)
- could not initialize class (58)
- beanpropertyrowmapper (58)
- word空格下划线不显示 (73)
- jar文件 (60)
- jsp内置对象 (58)
- makefile编写规则 (58)