java DelayQueue 关联redis

2024-05-29 05:08
文章标签 java redis 关联 delayqueue

本文主要是介绍java DelayQueue 关联redis,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

关联redis的DelayQueue

 

java 的延时队列相信很多人都用过,因为这东西应用场景太多了,但是有一点,就是如果重启服务器,那么延时队列里的东西都没了,特别是超多数据量的延时队列,这就是个大问题。

 

因此,有很多解决方案,比如出现了所谓的"redis延时队列",但是此延时队列并非我们想要的,因为他还是需要一个定时任务来执行,而我们真正想要的是这么个队列,有任务的时候,立刻以最快的速度取出,没有任务的时候线程阻塞,不额外占用cpu时间,做到系统负责应答,而非主动查询!!这点很关键,特别是高并发,大量数据的情况下系统负载又很大,这种方式就更加有优势了!!!

 

说了这么多,简单说一下思路,那就是用空间换取系统的效率。

 

思路是这样实现的还是基于java的延时队列

我们将收到的消息转化成自定义的消息对象(DelayQueueObject),放入延时队列,然后将他们在redis中进行备份,当延时队列中取出元素的时候,在redis中删除对应的key,需要删除某个元素的时候,用key从redis中取出对象,然后去延时队列中删除,中间用一些手段保证redis和延时队列中的元素一直统一,可以借鉴数据库的事务实现的思路。

但是!!!

需要重写DelayQueueObject.equals()方法,因为从redis中取出来的对象,并非是队列中的对象,所以默认的equals()是直接判断对象的地址的,不是同一个对象,当然不相等,队列中没有redis中的对象,所以删除会失败的!!!

而delayQueue.remove()方法调用的是PriorityQueue.remove(),而PriorityQueue.remove()方法又是利用对象的.equals()方法进行对象相等的比较而取出对象在队列中的位置,然后移除的,因此只要重写DelayQueueObject的.equals()方法即可让反序列化的对象和队列中的对象比较是否相等,具体.equals()怎么写,还需要看其中的元素是什么,找到对应的方法即可。

 

当然最好的办法还是自己封装一个延时队列,将操作合并在延时队列里面,这样用起来很方便,当然我的这种处理方式也还不错,如果大家不喜欢也不要喷我哈,欢迎大家讨论。

 

我这个人就是不爱多说话,希望谅解,直接上代码,不懂的私聊

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;import com.fasterxml.jackson.annotation.JsonIgnoreProperties;import net.sf.json.JSONArray;/*** 用于延时队列的对象* * @author TY** @param <E> 队列中存放的数据类型*/public class DelayQueueObject<E> implements Delayed, Serializable {/*** */private static final long serialVersionUID = 1L;private String id;private E object;private long delayTime;public DelayQueueObject() {}public DelayQueueObject(String id, E object, long delayTime) {this.id = id;this.object = object;this.delayTime = TimeUnit.NANOSECONDS.convert(delayTime, TimeUnit.MILLISECONDS) + System.nanoTime();}/*** 设置变形后的DelayTime* * @param delayTime*/public void setDecoratedDelayTime(long delayTime) {this.delayTime = delayTime;}/*** 从redis中取出后转换成对象使用* * @param delayQueueObject* @return* @throws Exception*/public static <E> String serializeToString(DelayQueueObject<E> delayQueueObject) throws Exception {ByteArrayOutputStream byteOut = new ByteArrayOutputStream();ObjectOutputStream objOut = new ObjectOutputStream(byteOut);objOut.writeObject(delayQueueObject);// 此处只能是ISO-8859-1,但是不会影响中文使用String objectStr = byteOut.toString("ISO-8859-1");return objectStr;}/*** 将对象存入reids中使用* * @param delayQueueObjectStr* @return* @throws Exception*/@SuppressWarnings("unchecked")public static <E> DelayQueueObject<E> deserializeToObject(String delayQueueObjectStr) throws Exception {// 判断参数if (delayQueueObjectStr == null || "".equals(delayQueueObjectStr)) {return null;}ByteArrayInputStream byteIn = new ByteArrayInputStream(delayQueueObjectStr.getBytes("ISO-8859-1"));ObjectInputStream objIn = new ObjectInputStream(byteIn);return (DelayQueueObject<E>) objIn.readObject();}/** 用于从delayQueue队列中取出元素时使用 */@Overridepublic boolean equals(Object obj) {if (this == obj) {return true;}if (obj == null) {return false;}if (getClass() != obj.getClass()) {return false;}@SuppressWarnings("unchecked")DelayQueueObject<E> other = (DelayQueueObject<E>) obj;if (delayTime != other.delayTime) {return false;}if (id == null) {if (other.id != null) {return false;}} else if (!id.equals(other.id)) {return false;}if (object == null) {if (other.object != null) {return false;}} else if (!object.equals(other.object)) {return false;}return true;}@Overridepublic String toString() {String objectStr;objectStr = JSONArray.fromObject(object).toString();return "{\"id\":" + id + ", \"object\":" + objectStr + ",\"delayTime\":" + delayTime + "}";}@Overridepublic int compareTo(Delayed delayed) {return (int) (this.getDelay(TimeUnit.MILLISECONDS) - delayed.getDelay(TimeUnit.MILLISECONDS));}@Overridepublic long getDelay(TimeUnit unit) {return unit.convert(this.delayTime - System.nanoTime(), TimeUnit.NANOSECONDS);}public String getId() {return id;}public void setId(String id) {this.id = id;}public E getObject() {return object;}public void setObject(E object) {this.object = object;}public long getDelayTime() {return delayTime;}public void setDelayTime(long delayTime) {this.delayTime = TimeUnit.NANOSECONDS.convert(delayTime, TimeUnit.MILLISECONDS) + System.nanoTime();}public static long getSerialversionuid() {return serialVersionUID;}}
/*** * 采用Redis备份恢复机制* * Component默认单利,完美!!* * @author TY 2018.09.21* */
@Component
public class OfflineDelayQueue {/** 延时消息队列map,用于对应delayObjectQueue */@Autowiredprivate RedisTemplate redisTemplate;/** 延时消息队列,用于处理offline */private DelayQueue<DelayQueueObject<List<String>>> delayQueue;/*** 启动线程,随系统启动,从reids初始化OfflineDelayQueue,用于加载系统重启前的OfflineDelayQueue* * @throws Exception*/@PostConstructprivate void initOfflineDelayQueue() throws Exception {delayQueue = new DelayQueue<DelayQueueObject<List<String>>>();Set<String> keys = redisTemplate.keys(Const.REDIS_OFFLINE_DELAY_QUEUE_KEY_HEAD + "*");Iterator<String> key = keys.iterator();while (key.hasNext()) {DelayQueueObject<List<String>> tempDelayQueueObject = DelayQueueObject.deserializeToObject(redisTemplate.get(key.next()));delayQueue.offer(tempDelayQueueObject);}}/*** 将delayQueueObject加入延时队列* * @param delayQueueObject* @throws Exception*/public void addToDelayQueue(DelayQueueObject<List<String>> delayQueueObject) throws Exception {// 如果延时队列添加报错,则抛出异常,上层处理delayQueue.offer(delayQueueObject);try {redisTemplate.set(Const.REDIS_OFFLINE_DELAY_QUEUE_KEY_HEAD + delayQueueObject.getId(),Const.REDIS_OFFLINE_DELAY_QUEUE_ACTIVE, DelayQueueObject.serializeToString(delayQueueObject));} catch (Exception e) {// 如果redis添加报错,则将队列元素删除,并向上层抛出异常delayQueue.remove(delayQueueObject);throw new Exception();}}/*** 根据id从延时队列中删除元素* * @param id* @return* @throws Exception*/public Boolean removeFromDelayQueue(String id) throws Exception {Boolean back = false;// redis中id对应的队列中的元素DelayQueueObject<List<String>> tempDelayQueueObject = DelayQueueObject.deserializeToObject(redisTemplate.get(Const.REDIS_OFFLINE_DELAY_QUEUE_KEY_HEAD + id));if (tempDelayQueueObject != null) {// 如果队列移除报报错,直接抛出异常,上层处理back = delayQueue.remove(tempDelayQueueObject);Boolean flag = false;try {// redis中没有相应元素不会抛异常flag = redisTemplate.del(Const.REDIS_OFFLINE_DELAY_QUEUE_KEY_HEAD + id);} catch (Exception e) {if (flag == false) {// 如果reids移除报错,再将元素加回去,并向上层抛出异常delayQueue.offer(tempDelayQueueObject);throw new Exception();}}}return back;}/*** 从队列中take元素* * @return* @throws Exception*/public DelayQueueObject<List<String>> takeFromDelayQueue() throws Exception {// 如果队列移除报报错,直接抛出异常,上层处理DelayQueueObject<List<String>> delayQueueObject = delayQueue.take();Boolean flag = false;try {flag = redisTemplate.del(Const.REDIS_OFFLINE_DELAY_QUEUE_KEY_HEAD + delayQueueObject.getId());} catch (Exception e) {if (flag == false) {// 如果map移除报错,再将元素加回去,并向上层抛出异常delayQueue.offer(delayQueueObject);throw new Exception();}}return delayQueueObject;}}
/*** DelayQueueConsumer* * @author TY 2018.09.21**/
@Component
public class DelayQueueConsumer implements Runnable {@Autowiredprivate OfflineDelayQueue offlineDelayQueue;/** 启动线程,随系统启动 */@PostConstructprivate void StartDelayQueueConsumer() {new Thread(this).start();}@Overridepublic void run() {System.out.println("---------------------------OfflineDelayQueue检测线程启动---------------------------");while (offlineDelayQueue != null && !Thread.currentThread().isInterrupted()) {try {DelayQueueObject<List<String>> delayQueueObject = offlineDelayQueue.takeFromDelayQueue();//自己发挥哦。。。//} catch (Exception e) {e.printStackTrace();// 防止死循环try {// 休眠10秒钟Thread.sleep(10000);} catch (InterruptedException e1) {e1.printStackTrace();}}}}
}

 

这篇关于java DelayQueue 关联redis的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1012728

相关文章

JVM 的类初始化机制

前言 当你在 Java 程序中new对象时,有没有考虑过 JVM 是如何把静态的字节码(byte code)转化为运行时对象的呢,这个问题看似简单,但清楚的同学相信也不会太多,这篇文章首先介绍 JVM 类初始化的机制,然后给出几个易出错的实例来分析,帮助大家更好理解这个知识点。 JVM 将字节码转化为运行时对象分为三个阶段,分别是:loading 、Linking、initialization

Spring Security 基于表达式的权限控制

前言 spring security 3.0已经可以使用spring el表达式来控制授权,允许在表达式中使用复杂的布尔逻辑来控制访问的权限。 常见的表达式 Spring Security可用表达式对象的基类是SecurityExpressionRoot。 表达式描述hasRole([role])用户拥有制定的角色时返回true (Spring security默认会带有ROLE_前缀),去

浅析Spring Security认证过程

类图 为了方便理解Spring Security认证流程,特意画了如下的类图,包含相关的核心认证类 概述 核心验证器 AuthenticationManager 该对象提供了认证方法的入口,接收一个Authentiaton对象作为参数; public interface AuthenticationManager {Authentication authenticate(Authenti

Spring Security--Architecture Overview

1 核心组件 这一节主要介绍一些在Spring Security中常见且核心的Java类,它们之间的依赖,构建起了整个框架。想要理解整个架构,最起码得对这些类眼熟。 1.1 SecurityContextHolder SecurityContextHolder用于存储安全上下文(security context)的信息。当前操作的用户是谁,该用户是否已经被认证,他拥有哪些角色权限…这些都被保

Spring Security基于数据库验证流程详解

Spring Security 校验流程图 相关解释说明(认真看哦) AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证,则会调用 #attemptAuthentica

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门(一):登录与退出》 《Spring-Security-入门(二):基于数据库验证》 《Spring-Security-入门(三):密码加密》 《Spring-Security-入门(四):自定义-Filter》 《Spring-Security-入门(五):在 Sprin

Java架构师知识体认识

源码分析 常用设计模式 Proxy代理模式Factory工厂模式Singleton单例模式Delegate委派模式Strategy策略模式Prototype原型模式Template模板模式 Spring5 beans 接口实例化代理Bean操作 Context Ioc容器设计原理及高级特性Aop设计原理Factorybean与Beanfactory Transaction 声明式事物

Java进阶13讲__第12讲_1/2

多线程、线程池 1.  线程概念 1.1  什么是线程 1.2  线程的好处 2.   创建线程的三种方式 注意事项 2.1  继承Thread类 2.1.1 认识  2.1.2  编码实现  package cn.hdc.oop10.Thread;import org.slf4j.Logger;import org.slf4j.LoggerFactory

JAVA智听未来一站式有声阅读平台听书系统小程序源码

智听未来,一站式有声阅读平台听书系统 🌟&nbsp;开篇:遇见未来,从“智听”开始 在这个快节奏的时代,你是否渴望在忙碌的间隙,找到一片属于自己的宁静角落?是否梦想着能随时随地,沉浸在知识的海洋,或是故事的奇幻世界里?今天,就让我带你一起探索“智听未来”——这一站式有声阅读平台听书系统,它正悄悄改变着我们的阅读方式,让未来触手可及! 📚&nbsp;第一站:海量资源,应有尽有 走进“智听

零基础学习Redis(10) -- zset类型命令使用

zset是有序集合,内部除了存储元素外,还会存储一个score,存储在zset中的元素会按照score的大小升序排列,不同元素的score可以重复,score相同的元素会按照元素的字典序排列。 1. zset常用命令 1.1 zadd  zadd key [NX | XX] [GT | LT]   [CH] [INCR] score member [score member ...]