阻塞队列之七:DelayQueue延时队列
一、DelayQueue简介
是一个无界的BlockingQueue,用于放置实现了Delayed接口的对象,其中的对象只能在其到期时才能从队列中取走。这种队列是有序的(PriorityQueue实际存放Delayed接口对象),即队头对象的延迟到期时间最短(队列顶端总是最小的元素)。注意:不能将null元素放置到这种队列中。
DelayQueue在poll/take的时候,队列中元素会判定这个elment有没有达到超时时间,如果没有达到,poll返回null,而take进入等待状态。但是,除了这两个方法,队列中的元素会被当做正常的元素来对待。例如,size方法返回所有元素的数量,而不管它们有没有达到超时时间。而协调的Condition available只对take和poll是有意义的。
二、DelayQueue源码分析
2.1、DelayQueue的lock
DelayQueue使用一个可重入锁和这个锁生成的一个条件对象进行并发控制。
private final transient ReentrantLock lock = new ReentrantLock(); //内部用于存储对象private final PriorityQueue<E> q = new PriorityQueue<E>();/*** Thread designated to wait for the element at the head of* the queue. This variant of the Leader-Follower pattern* (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to* minimize unnecessary timed waiting. When a thread becomes* the leader, it waits only for the next delay to elapse, but* other threads await indefinitely. The leader thread must* signal some other thread before returning from take() or* poll(...), unless some other thread becomes leader in the* interim. Whenever the head of the queue is replaced with* an element with an earlier expiration time, the leader* field is invalidated by being reset to null, and some* waiting thread, but not necessarily the current leader, is* signalled. So waiting threads must be prepared to acquire* and lose leadership while waiting.*/private Thread leader = null;/*** Condition signalled when a newer element becomes available* at the head of the queue or a new thread may need to* become leader.*/private final Condition available = lock.newCondition();
2.2、成员变量
要先了解下DelayQueue中用到的几个关键对象:
2.2.1、Delayed, 一种混合风格的接口,用来标记那些应该在给定延迟时间之后执行的对象。
此接口的实现必须定义一个 compareTo()方法,该方法提供与此接口的 getDelay()方法一致的排序。
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>implements BlockingQueue<E> {
DelayQueue是一个BlockingQueue,其泛型类的参数是Delayed接口对象。
Delayed接口:
public interface Delayed extends Comparable<Delayed> {long getDelay(TimeUnit unit); //返回与此对象相关的剩余延迟时间,以给定的时间单位表示。
}
Comparable接口:
public interface Comparable<T> {public int compareTo(T o); }
Delayed扩展了Comparable接口,比较的基准为延时的时间值,Delayed接口的实现类getDelay的返回值应为固定值(final)。
2.2.2、PriorityQueue,优先级队列存放有序对象
优先队列的比较基准值是时间。详解见《阻塞队列之八:PriorityBlockingQueue优先队列》
DelayQueue的关键元素BlockingQueue、PriorityQueue、Delayed。可以这么说,DelayQueue是一个使用优先队列(PriorityQueue)实现的BlockingQueue,优先队列的比较基准值是时间。
public class DelayQueue<E extends Delayed> implements BlockingQueue<E> { private final PriorityQueue<E> q = new PriorityQueue<E>(); }
总结:DelayQueue内部是使用PriorityQueue实现的,DelayQueue = BlockingQueue + PriorityQueue + Delayed。
2.3、构造函数
public DelayQueue() {} public DelayQueue(Collection<? extends E> c) {this.addAll(c);}
public boolean offer(E e, long timeout, TimeUnit unit) {return offer(e);}
超时的参数被忽略,因为是无界的。不会阻塞或超时。
2.4、入队
public boolean add(E e) {return offer(e);}public void put(E e) {offer(e);}public boolean offer(E e) {final ReentrantLock lock = this.lock;lock.lock();try {q.offer(e);if (q.peek() == e) {//添加元素后peek还是e,重置leader,通知条件队列 leader = null;available.signal();}return true;} finally {lock.unlock();}}
2.5、出队
public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { E first = q.peek(); if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0) //队列为空或者延迟时间未过期 return null; else return q.poll(); } finally { lock.unlock(); } } /** * take元素,元素未过期需要阻塞 */ public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { for (;;) { E first = q.peek(); if (first == null) available.await(); //队列空,加入条件队列 else { long delay = first.getDelay(TimeUnit.NANOSECONDS); //获取剩余延迟时间 if (delay <= 0) //小于0,那就poll元素 return q.poll(); else if (leader != null) //有延迟,检查leader,不为空说明有其他线程在等待,那就加入条件队列 available.await(); else { Thread thisThread = Thread.currentThread(); leader = thisThread; //设置当前为leader等待 try { available.awaitNanos(delay); //条件队列等待指定时间 } finally { if (leader == thisThread) //检查是否被其他线程改变,没有就重置,再次循环 leader = null; } } } } } finally { if (leader == null && q.peek() != null) //leader为空并且队列不空,说明没有其他线程在等待,那就通知条件队列 available.signal(); lock.unlock(); } } /** * 响应超时的poll */ public E poll(long timeout, TimeUnit unit) throws InterruptedException { long nanos = unit.toNanos(timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { for (;;) { E first = q.peek(); if (first == null) { if (nanos <= 0) return null; else nanos = available.awaitNanos(nanos); } else { long delay = first.getDelay(TimeUnit.NANOSECONDS); if (delay <= 0) return q.poll(); if (nanos <= 0) return null; if (nanos < delay || leader != null) nanos = available.awaitNanos(nanos); else { Thread thisThread = Thread.currentThread(); leader = thisThread; try { long timeLeft = available.awaitNanos(delay); nanos -= delay - timeLeft; } finally { if (leader == thisThread) leader = null; } } } } } finally { if (leader == null && q.peek() != null) available.signal(); lock.unlock(); } } /** * 获取queue[0],peek是不移除的 */ public E peek() { final ReentrantLock lock = this.lock; lock.lock(); try { return q.peek(); } finally { lock.unlock(); } }
三、JDK或开源框架中使用
ScheduledThreadPoolExecutor中使用了DelayedWorkQueue。
应用场景
下面的应用场景是来源于网上,虽然借用DelayedQueue可以快速找到要“失效”的对象,但DelayedQueue内部的PriorityQueue的(插入、删除时的排序)也耗费资源。
a) 关闭空闲连接。服务器中,有很多客户端的连接,空闲一段时间之后需要关闭之。
b) 缓存。缓存中的对象,超过了空闲时间,需要从缓存中移出。
c) 任务超时处理。在网络协议滑动窗口请求应答式交互时,处理超时未响应的请求。
d)session超时管理,网络应答通讯协议的请求超时处理。
四、示例
1、缓存示例
- 当向缓存中添加key-value对时,如果这个key在缓存中存在并且还没有过期,需要用这个key对应的新过期时间
- 为了能够让DelayQueue将其已保存的key删除,需要重写实现Delayed接口添加到DelayQueue的DelayedItem的hashCode函数和equals函数
- 当缓存关闭,监控程序也应关闭,因而监控线程应当用守护线程
以下是Sample,是一个缓存的简单实现。共包括三个类Pair、DelayItem、Cache。如下:
package com.dxz.concurrent.delayqueue;public class Pair<K, V> {public K key;public V value;public Pair() {}public Pair(K first, V second) {this.key = first;this.value = second;}@Overridepublic String toString() {return "Pair [key=" + key + ", value=" + value + "]";}}
以下是Delayed的实现
package com.dxz.concurrent.delayqueue;import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong;public class DelayItem<T> implements Delayed {/** Base of nanosecond timings, to avoid wrapping */private static final long NANO_ORIGIN = System.nanoTime();/*** Returns nanosecond time offset by origin*/final static long now() {return System.nanoTime() - NANO_ORIGIN;}/*** Sequence number to break scheduling ties, and in turn to guarantee FIFO* order among tied entries.*/private static final AtomicLong sequencer = new AtomicLong(0);/** Sequence number to break ties FIFO */private final long sequenceNumber;/** The time the task is enabled to execute in nanoTime units */private final long time;private final T item;public DelayItem(T submit, long timeout) {this.time = now() + timeout;this.item = submit;this.sequenceNumber = sequencer.getAndIncrement();}public T getItem() {return this.item;}public long getDelay(TimeUnit unit) {long d = unit.convert(time - now(), TimeUnit.NANOSECONDS);return d;}public int compareTo(Delayed other) {if (other == this) // compare zero ONLY if same objectreturn 0;if (other instanceof DelayItem) {DelayItem x = (DelayItem) other;long diff = time - x.time;if (diff < 0)return -1;else if (diff > 0)return 1;else if (sequenceNumber < x.sequenceNumber)return -1;elsereturn 1;}long d = (getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS));return (d == 0) ? 0 : ((d < 0) ? -1 : 1);} }
以下是Cache的实现,包括了put和get方法,还包括了可执行的main函数。
package com.dxz.concurrent.delayqueue;import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.DelayQueue; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger;public class Cache<K, V> {private static final Logger LOG = Logger.getLogger(Cache.class.getName());private ConcurrentMap<K, V> cacheObjMap = new ConcurrentHashMap<K, V>();private DelayQueue<DelayItem<Pair<K, V>>> q = new DelayQueue<DelayItem<Pair<K, V>>>();private Thread daemonThread;public Cache() {Runnable daemonTask = new Runnable() {public void run() {daemonCheck();}};daemonThread = new Thread(daemonTask);daemonThread.setDaemon(true);daemonThread.setName("Cache Daemon");daemonThread.start();}private void daemonCheck() {if (LOG.isLoggable(Level.INFO))LOG.info("cache service started.");for (;;) {try {DelayItem<Pair<K, V>> delayItem = q.take();if (delayItem != null) {// 超时对象处理Pair<K, V> pair = delayItem.getItem();cacheObjMap.remove(pair.key, pair.value); // compare and// remove }} catch (InterruptedException e) {if (LOG.isLoggable(Level.SEVERE))LOG.log(Level.SEVERE, e.getMessage(), e);break;}}if (LOG.isLoggable(Level.INFO))LOG.info("cache service stopped.");}// 添加缓存对象public void put(K key, V value, long time, TimeUnit unit) {V oldValue = cacheObjMap.put(key, value);if (oldValue != null) {boolean result = q.remove(new DelayItem<Pair<K, V>>(new Pair<K, V>(key, oldValue), 0L));System.out.println("remove:="+result);}long nanoTime = TimeUnit.NANOSECONDS.convert(time, unit);q.put(new DelayItem<Pair<K, V>>(new Pair<K, V>(key, value), nanoTime));}public V get(K key) {return cacheObjMap.get(key);}public DelayQueue<DelayItem<Pair<K, V>>> getQ() {return q;}public void setQ(DelayQueue<DelayItem<Pair<K, V>>> q) {this.q = q;}// 测试入口函数public static void main(String[] args) throws Exception {Cache<Integer, String> cache = new Cache<Integer, String>();cache.put(1, "aaaa", 60, TimeUnit.SECONDS);cache.put(1, "aaaa", 10, TimeUnit.SECONDS);//cache.put(1, "ccc", 60, TimeUnit.SECONDS);cache.put(2, "bbbb", 30, TimeUnit.SECONDS);cache.put(3, "cccc", 66, TimeUnit.SECONDS);cache.put(4, "dddd", 54, TimeUnit.SECONDS);cache.put(5, "eeee", 35, TimeUnit.SECONDS);cache.put(6, "ffff", 38, TimeUnit.SECONDS);cache.put(1, "aaaa", 70, TimeUnit.SECONDS);for(;;) {Thread.sleep(1000 * 2);{for(Object obj : cache.getQ().toArray()) {System.out.print(((DelayItem)obj).toString());System.out.println(",");}System.out.println();}}} }
结果片段1:(重复key的Delayed对象将从DelayedQueue中移除)
remove:=true remove:=true 七月 04, 2017 11:28:36 上午 com.dxz.concurrent.delayqueue.Cache daemonCheck 信息: cache service started. DelayItem [sequenceNumber=3, time=30000790187, item=Pair [key=2, value=bbbb]], DelayItem [sequenceNumber=6, time=35000842411, item=Pair [key=5, value=eeee]], DelayItem [sequenceNumber=7, time=38000847189, item=Pair [key=6, value=ffff]], DelayItem [sequenceNumber=5, time=54000835925, item=Pair [key=4, value=dddd]], DelayItem [sequenceNumber=4, time=66000803499, item=Pair [key=3, value=cccc]], DelayItem [sequenceNumber=9, time=70000900437, item=Pair [key=1, value=aaaa]],
结果片段2:(队头对象将最先过时,可以被take()出来,这段代码在daemonCheck()方法中,即对超时对象的处理,如这里是清理session集合对象)
... DelayItem [sequenceNumber=3, time=30000665600, item=Pair [key=2, value=bbbb]], DelayItem [sequenceNumber=6, time=35000689152, item=Pair [key=5, value=eeee]], DelayItem [sequenceNumber=7, time=38000694272, item=Pair [key=6, value=ffff]], DelayItem [sequenceNumber=5, time=54000685398, item=Pair [key=4, value=dddd]], DelayItem [sequenceNumber=4, time=66000679595, item=Pair [key=3, value=cccc]], DelayItem [sequenceNumber=9, time=70000728406, item=Pair [key=1, value=aaaa]],DelayItem [sequenceNumber=6, time=35000689152, item=Pair [key=5, value=eeee]], DelayItem [sequenceNumber=5, time=54000685398, item=Pair [key=4, value=dddd]], DelayItem [sequenceNumber=7, time=38000694272, item=Pair [key=6, value=ffff]], DelayItem [sequenceNumber=9, time=70000728406, item=Pair [key=1, value=aaaa]], DelayItem [sequenceNumber=4, time=66000679595, item=Pair [key=3, value=cccc]], ...
阻塞队列之七:DelayQueue延时队列相关推荐
- rabbitmq安装延时队列插件实现延时队列
前言 这个很难查看消息堆积的情况,因为他把要发送的延时消息存在本地的分布式mnesia数据库中,其次过期时间为最大int值,超过这个值(大概49天)得代码判定重复过期设置. 下载插件地址 要注意和自己 ...
- 一口气说出 6种 延时队列的实现方案,大厂offer稳稳的
下边会介绍多种实现延时队列的思路,文末提供有几种实现方式的 github地址.其实哪种方式都没有绝对的好与坏,只是看把它用在什么业务场景中,技术这东西没有最好的只有最合适的. 一.延时队列的应用 什么 ...
- 一口气说出 6 种延时队列的实现方法,面试官满意的笑了
这是我的第 193 期分享 作者 | 程序员内点事 来源 | 程序员内点事(ID:chegnxy-nds) 分享 | Java中文社群(ID:javacn666) 五一期间原计划是写两篇文章,看一本技 ...
- 一口气说出 6种 延时队列的实现方法,面试官也得服
一.延时队列的应用 什么是延时队列?顾名思义:首先它要具有队列的特性,再给它附加一个延迟消费队列消息的功能,也就是说可以指定队列中的消息在哪个时间点被消费. 延时队列在项目中的应用还是比较多的,尤其像 ...
- 【java】6个延时队列的实现方案
1.概述 转载:6个延时队列的实现方案 [编者的话]个人比较喜欢一些实践类的东西,既学习到知识又能让技术落地,能搞出个demo最好,本来不知道该分享什么主题,好在最近项目紧急招人中,而我有幸做了回面试 ...
- 【Redis核心原理和应用实践】应用 2:缓兵之计 —— 延时队列
我们平时习惯于使用 Rabbitmq 和 Kafka 作为消息队列中间件,来给应用程序之间增加异步消息传递功能.这两个中间件都是专业的消息队列中间件,特性之多超出了大多数人的理解能力. 使用过 Ra ...
- 应用 2:缓兵之计 ——延时队列
1.redis做消息队列 使用list(列表) 数据结构常用来作为异步消息队列使用,使用rpush/lpush操作入队列, 使用 lpop 和 rpop来出队列 2.队列空了怎么办 客户端是通过队列的 ...
- 延时队列的几种实现方式
延时队列的几种实现方式 何为延迟队列? 顾名思义,首先它要具有队列的特性,再给它附加一个延迟消费队列消息的功能,也就是说可以指定队列中的消息在哪个时间点被消费. 延时队列能做什么? 延时队列多用于需要 ...
- RabbitMQ真延时队列实现消息提醒功能
RabbitMQ真延时队列实现消息提醒功能 一.需求场景 用户可以制定多个计划,同时可给该计划设置是否需要到点提醒,且中途可以取消提醒或修改提醒时间. 二.需要解决的问题 学习过rabbitmq的同学 ...
最新文章
- 基于S3C4510B的一个简单BSP的开发报告
- 大型网站架构系列:缓存在分布式系统中的应用(二)
- PHP框架CodeIgniter之连接MS Sqlserver2014及URL Rewrite问题解决
- http://blog.csdn.net/chrisniu1984/article/details/12050951
- python列表常用操作函数_Python入门——列表常用操作
- 代码编辑器VS Code的“Chromium”版来啦:安全、开源、保护你的隐私
- 2019-12-02 调用C++高精度时钟 std::chrono::high_resolution_clock的方法
- 超神学院暗物质计算机,超神学院之进击的赛亚人
- 【黑马Python】(3)
- 读书笔记 大前研一 《M型社会》
- Android广告图片轮播控件banner
- Nginx证书配置:cer文件和jks文件转nginx证书.crt和key文件
- 差分 + 差分矩阵 (差分)
- Linux C alarm的使用
- NETCONF配置CISCO XE(csr1000v)初体验
- C++ 利用 windbg + dump + map + cod 文件分析 crash 原因
- 宽依赖和窄依赖_Spark宽依赖和窄依赖深度剖析
- JAVA 接口 验证哥特巴赫猜想
- 旋转编码器(STM32)
- 毕业生必看的方法:CAJ文件如何免费转换成word文档
热门文章
- mysql 常用命令 汇总
- 实验室里人越来越少啊!
- android 6.0 api 管理,Android 6.0(API23)权限申请问题
- 速读训练软件_记忆宫殿记忆力训练教程-第八天
- python处理excel的方法有哪些_python简单处理excel方法
- java8中Predicate用法
- java中break和continue的用法例子
- 数据库:SQLServer 实现行转列、列转行用法笔记
- 【原创】MVC+ZTree实现权限树的功能
- ds证据理论python实现_ALI模型理论以及Python实现