java DelayQueue 关联redis

客官°小女子只卖身不卖艺 2022-05-10 01:34 137阅读 0赞

# 关联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队列中取出元素时使用 */
    	@Override
    	public 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;
    	}
    
    	@Override
    	public String toString() {
    		String objectStr;
    		objectStr = JSONArray.fromObject(object).toString();
    		return "{\"id\":" + id + ", \"object\":" + objectStr + ",\"delayTime\":" + delayTime + "}";
    	}
    
    	@Override
    	public int compareTo(Delayed delayed) {
    		return (int) (this.getDelay(TimeUnit.MILLISECONDS) - delayed.getDelay(TimeUnit.MILLISECONDS));
    	}
    
    	@Override
    	public 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 */
    	@Autowired
    	private RedisTemplate redisTemplate;
    
    	/** 延时消息队列，用于处理offline */
    	private DelayQueue<DelayQueueObject<List<String>>> delayQueue;
    
    	/**
    	 * 启动线程,随系统启动,从reids初始化OfflineDelayQueue，用于加载系统重启前的OfflineDelayQueue
    	 * 
    	 * @throws Exception
    	 */
    	@PostConstruct
    	private 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 {
    
    	@Autowired
    	private OfflineDelayQueue offlineDelayQueue;
    
    	/** 启动线程,随系统启动 */
    	@PostConstruct
    	private void StartDelayQueueConsumer() {
    		new Thread(this).start();
    	}
    
    	@Override
    	public 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();
    				}
    			}
    		}
    	}
    }