DelayQueue 是一个无界延时阻塞队列,元素顺序按照过期时间排序,通过显式锁 ReentrantLock 保证并发安全,队列中的存储的元素必须实现 Delayed 接口,也就是说只允许放入可以“延期”的元素。获取元素时,只有当元素期满之后才可获取。DelayQueue 相对比较简单,不过在我们之后线程池源码分析中会遇到,所以这里也简单介绍一下。
概述
DelayQueue 的延时策略可以总结为以下几点:
- 存储元素必须实现
Delayed
接口 - 内部持有一个
ReentrantLock
保证线程安全 - 使用优先级队列
PriorityQueue
实现元素存储 - 持有一个优化内部阻塞通知的线程
leader
- 用于实现阻塞的
Condition
对象
关于 Delayed 和 PriorityQueue:
- Delayed 是一个具有过期时间的元素类型
- PriorityQueue 是二叉堆实现的根据队列里元素的某些属性排序的的优先级队列,内部持有一个比较器comparator(参考JUC源码分析-集合篇(七):PriorityBlockingQueue)
DelayQueue 其实就是在每次往优先级队列中添加元素,然后以元素的 delay(过期值)作为排序的因素,以此来达到先过期的元素会拍在队首,每次从队列里取出来都是最先要过期的元素。
数据结构及核心参数
//锁
private final transient ReentrantLock lock = new ReentrantLock();
//内部使用PriorityQueue存储元素
private final PriorityQueue<E> q = new PriorityQueue<E>();
//等待获取队列头元素的线程
private Thread leader = null;
//当一个新任务在队列的头部可用,或者新线程可能需要成为leader时,唤醒等待条件
private final Condition available = lock.newCondition();
这里我们主要介绍一下leader
变量:
leader
是等待获取队列头元素的线程,应用主从式设计减少不必要的等待。当一个线程为leader
,它只会等待下一个延迟届期,但是其他线程的等待是不确定的。在从take()
或poll()
获取数据返回前,leader
线程必须唤醒其他等待的线程,除非其他线程在这期间变成leader
。如果队列头被一个有着更快过期时间的元素替换掉,leader
将会被设置为null
而失效,并唤醒其他等待线程(不一定是当前leader
线程)。所以等待线程在等待期间必须时刻准备获取或失去 leader 权限。
源码解析
offer(E e)
DelayQueue 添加或入列操作方法包括put、add、offer
,都是通过offer
方法实现,所以我们这里只对offer
进行解析:
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
//调用priorityQueue的offer出列
q.offer(e);
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
说明:首先执行锁操作,把元素添加到优先级队列priorityQueue
中;如果当前元素是队列的头元素,则设置leader
为空,并唤醒所有等待available
的线程;最后释放锁。
这里用到了PriorityQueue
的offer()
和peek()
方法。PriorityQueue 的入列操作与 PriorityBlockingQueue 基本一致,这里大家可以参考笔者的另一篇文章JUC源码分析-集合篇(七):PriorityBlockingQueue,查看 PriorityBlockingQueue 的入列方式,本篇就不多赘述了。
take()
DelayQueue 的出列或获取元素方法包括poll、take、peek
,poll
直接获取元素,如果队列中没有届期元素返回null;take
会一直等待元素可用;peek
只获取但不移除元素。相对来说,take
方法内包括了另外两个方法的逻辑,所以这里我们只针对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(NANOSECONDS);//延时时间
if (delay <= 0)
return q.poll();
//释放first的引用,避免内存泄漏
first = null; // don't retain ref while waiting
if (leader != null)//leader不为空,证明有其他线程已经获取到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();
}
}
说明:函数执行逻辑如下:
- 首先对队列加(响应中断)锁;
- 调用
priorityQueue.peek()
方法获取首节点; - 首节点为空则调用
available.await()
等待被唤醒; - 首节点不为空,并且延迟时间为0则执行出列操作;
- 首节点不为空但是尚未届期,则首先释放 first 的引用,防止内存泄漏,然后执行如下逻辑:
-
leader
不为空,证明有其他线程已经获取到leader
,当前线程加入等待条件队列; -
leader
为空,使leader
指向当前线程,等待节点的 delay 时间;
- 处理结束,如果队列中还有节点就唤醒在
available
条件上等待的线程。 - unlock
小结
如果你理解了我们前几章的内容,会发现 DelayQueue 还是比较简单的。
本章重点:DelayQueue 的延时策略。