1 前言
前两节我们看了BlockingQueue阻塞队列的两个子类，LinkedBlockingQueue、ArrayBlockingQueue，它们都是使用了ReentrantLock、Condition的来实现的，在进行插入操作、拉取数据操作之前为了并发安全都需要进行加锁；
然后插入时候在容量满的时候发现没有空间了，这时候调用Condition.await方法进行阻塞等待；等数据拉取线程拉取了元素，有位置可以插入的时候调用Condition.singal方法将插入线程唤醒，让插入线程继续。
同理对于数据拉取线程也是一样的，数据拉取线程拉取数据发现队列是空的，调用Condition.await方法阻塞等待；数据插入线程插入数据之后调用Condition.singal方法将阻塞的数据拉取线程唤醒，告诉它有数据可以拉取了。
我们本节继续研究阻塞队列，不过这节看的是延迟阻塞队列，对应的子类是DelayQueue。
2 DelayQueue是什么
延迟阻塞队列，是实现很多延迟任务调度，定时任务调度的基础，它大概功能有：
（1）首先这是一个队列，是用来进行存储元素，提供别人取的，具有存、取这两个基本的功能。
（2）其次，它具有延迟的功能也就是具有时效管理的功能；应该就是说我可以存入一个元素，这个元素指定一个延迟时间，然后延迟时间到达后就可以取出，时间没到之前是不能取出这个元素的。
比如说我添加一个元素 item，指定的延迟时间是1s，调用put方法放入阻塞队列中；这个时候我使用另外的一个线程立即调用take方法是取不到这个元素的，必须要等到1s之后，调用take方法才能取到这个元素，画个图来理解一下：
我们先浅浅的透露一下，元素是怎么具有延迟时间的，就是继承一个延迟时间的接口：
// 这里存入DelayQueue的元素类型使用泛型E来表示
// 其中泛型E必须继承Delayed接口
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E> {
存入DelayQueue延迟队列的元素类型E，必须继承Delayed接口，我们下面看看Delayed接口有啥：
// Delayed接口继承了Comparable接口，具有时间比较的功能
public interface Delayed extends Comparable<Delayed> {
// 有一个getDelay方法，获取延迟的时间是多少
long getDelay(TimeUnit unit);
}
从Delayed接口我们得到如下信息：
（1）具有一个方法getDelay获取当前元素的延迟时间是多少，这样每一个存入队列的元素E都具有延迟时间了
（2）Delayed接口继承了Comparable接口，具有比较的功能。
猜测是需要根据时间进行比较排序，延迟时间比较早的排在前面，时间大的排在后面；这样从延迟队列取元素的时候，先取延迟时间比较小的，然后才能取比较大的。
所以让存入的元素类型E都继承Delayed接口，必须是新getDelay方法，这样每个元素都具有延迟时间了，我们接下来就来看看DelayQueue内部原理。
3 DelayQueue内部源码
3.1 内部属性
我们先来看一下DelayQueue内部有哪些属性：
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E> {
// 锁，保证并发安全性，结合Condition实现阻塞、唤醒机制
private final transient ReentrantLock lock = new ReentrantLock();
// 优先级队列，这个队列就是存储延迟元素的容器
// 延迟元素E按照延迟时间作为优先级存储在PriorityQueue优先级队列中
// 延迟时间小的存放在队列前面，延迟时间大的存在队列后面
private final PriorityQueue<E> q = new PriorityQueue<E>();
// leader线程（第一个等待获取数据的线程）
private Thread leader = null;
// 等待条件，如果没有可取元素，则会调用available.await方法阻塞等待
private final Condition available = lock.newCondition();
}
构造方法比较简单我们就不贴了哈，针对这些属性我们简单看一下：
（1）首先DeleyQueue内部有一个PriorityQueue优先级队列来存储延迟元素，使用延迟时间作为优先级；延迟时间越小优先级越高，排在越前面，延迟时间越大优先级越低，排在越后面。
（2）有一个ReentrantLock互斥锁，获取数据之前需要进行加锁；还有一个available的Condition等待条件，如果当前PriorityQueue队列排在最前面的元素，延迟时间大于当前时间，当前没有可取元素，则available条件不满足，需要阻塞等待。
（3）leader表示当前第一个阻塞等待获取元素的线程，leader就是第一个，排头的意思。
3.2 add方法
public boolean add(E e) {
// 调用内部的offer(e)方法
return offer(e);
}
3.3 put方法
public void put(E e) {
// 调用内部的offer(e)方法
offer(e);
}
3.4 offer(E e, long timeout, TimeUnit unit)方法
public boolean offer(E e, long timeout, TimeUnit unit) {
// 调用内部的offer(e)方法
return offer(e);
}
这里看得出，插入都是依赖于内部的offer(e)方法，我们继续看下offer(E e)方法的内部源码：
3.5 offer(E e)方法
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
// 插入之前先获取锁
lock.lock();
try {
// 往优先级队列里面加入插入的元素
// 优先级队列会根据插入元素e的延迟时间，自动给你排序
q.offer(e);
if (q.peek() == e) {
// 如果队列的第一个元素是当前元素e，有两种可能
// 第一种：之前队列没有元素，可能有部分线程取元素的时候被阻塞了，此时需要唤醒阻塞线程
// 第二种：e的延迟时间最小，排在阻塞队列第一位，很快e就会被取出了
// 下面的available方法唤醒阻塞的线程，所以就不存在第一个阻塞的线程了，这里需要将leader设置为null
leader = null;
// 很快就有元素可以能取出了，唤醒阻塞的线程
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
我们来画个图理解一下：
上图大致的执行过程：
（1）插入元素的时候最终都是调用到offer(e)方法，首先老规矩，为了并发安全，都需要进行加锁
（2）加锁成功之后，将插入元素插入到优先级队列PriorityQueue中，队列会根据元素的延迟时间自动给你排序。
（3）插入成功之后，判断新插入的元素e 是不是延迟时间排在第一位的元素
（4）如果是，则可能之前队列是空的，或者新插入的元素延迟时间最小。有可能之前有别的线程取数据的时候被阻塞了，此时调用available.singal方法唤醒，提醒他们准备有数据可以获取了
（5）最后就是调用lock.unlock方法释放锁了
这里提一下，跟前两节我们看的LinkedBlockingQueue、ArrayBlockingQueue不同之处在于：插入的时候没有进行容量判断，直接就插入了。DelayQueue设计的初衷就是不会阻塞插入的，队列是没有容量限制的，是一个无界的队列。它这里阻塞队列主要指的是取元素的时候，如果队列是空的，或者有没有延迟时间小于当前时间的元素，取数据的线程就需要阻塞等待。
3.6 take方法
我们继续，看一下DelayQueue取数据的方法：
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
// 老规矩，先加锁
lock.lockInterruptibly();
try {
// 这里循环尝试获取
for (;;) {
// 获取优先级队列第一个元素，也就是延迟时间最靠前的元素
E first = q.peek();
// 如果当前队列是空，没有元素
if (first == null)
// 此时没有元素可以获取，需要阻塞等待一下，等有元素可取的时候别的线程唤醒告诉你
available.await();
else {
// 获取一下队列里面延迟时间最小的元素，还剩余的延迟时间是多少（单位是纳秒，NANOSECONDS是纳秒）
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
// 如果剩余延迟时间小于等于0，说明到时间了，可以取出了
return q.poll();
// 走到这里说明剩余延迟时间，delay > 0说明当前肯定没有到时间的元素，没有元素可取
// 设置first引用指向null
first = null; // don't retain ref while waiting
if (leader != null)
// 如果leader != null，说明在自己之前已经有线程取元素失败而阻塞等待了
available.await();
else {
// 走到这里说明leader == null，说明自己之前没有线程因为获取失败而阻塞等待
Thread thisThread = Thread.currentThread();
// 那么自己就是第一个获取失败的线程，leader就是自己，表示第一个获取失败而阻塞等待的线程
leader = thisThread;
try {
// 这里阻塞等待，最多阻塞delay时间，因为排在最前的元素剩余延迟时间为delay纳秒
// 所以大概delay纳秒之后就有元素可取了，所以最多需要等待delay纳秒时间
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
// 走到这里说明已经从阻塞等待中苏醒了，自己已经不是第一个阻塞等待的线程了
// 因为自己已经醒了嘛，所以不是阻塞的了，因此设置leader = null
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
// 走到这里说明自己取出了一个元素，同时q.peek() != null 队列里面可能还有元素可取
// 此时唤醒一下阻塞的线程，让它去尝试取元素
available.signal();
// 这里就是常规的释放锁了
lock.unlock();
}
}
我们接着上边的offer图，画一下take方法执行过程：
图中的蓝色为offer,紫色表示take方法的流程：
（1）线程B要调用take方法取一个元素，首先需要进行加锁
（2）加锁成功之后，从优先级队列中取出第一个元素，first = p.peek()
（3）如果first == null 说明队列一个元素都没有，此时就需要调用await方法阻塞等待了，但是阻塞之前先判断一下leader 是否为空，如果是空，说明自己是第一个阻塞等待线程，设置leader为自己
（4）如果first 非空，获取first元素的剩余延迟时间delay，如果delay <= 0 说明到时间了，可从队列取出这个元素
（5）如果delay > 0说明元素未到时间，最多还需要等待delay时间，同样需要阻塞，调用awaitNanos（delay）阻塞等待delay纳秒时间。同时也需要进行leader判断等
（6）最后到达finally时候，自己可能苏醒了，不再是阻塞状态，如果leader是自己，则需要设置leader = null，判断p.peek是否为空，也就是队列还是否有元素，如果有，则需要唤醒一下其它等待线程，让它们尝试再去获取元素。
3.7 poll方法
我们继续，看下其它取数的poll方法：
public E poll() {
final ReentrantLock lock = this.lock;
// 获取锁
lock.lock();
try {
// 获取优先级队列第一个元素
E first = q.peek();
// 如果第一个元素为空，说明队列根本没有元素
// 如果第一个元素剩余延迟时间大于0，说明还没到取出时间
if (first == null || first.getDelay(NANOSECONDS) > 0)
// 这里就直接返回null，注意这里不会阻塞
return null;
else
// 否则就从队列弹出这个元素
return q.poll();
} finally {
// 是否锁
lock.unlock();
}
}
这里poll方法的源码流程很简单了，就是直接取第一个元素，如果元素非空，并且剩余延迟时间小于等于0，则返回这个元素，否则返回空，图这里就不画了。
再来看下poll(long timeout, TimeUnit unit)方法源码：
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
// 最大阻塞时间转换为纳秒
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
// 进行加锁
lock.lockInterruptibly();
try {
// for循环尝试
for (;;) {
// 获取第一个元素
E first = q.peek();
if (first == null) {
// 如果第一个元素为null,并且剩余等待时间小于等于0，说明超时了，返回null
if (nanos <= 0)
return null;
else
// 否则这里就进行阻塞等待，最多阻塞nanos纳秒
nanos = available.awaitNanos(nanos);
} else {
// 如果first不为空，获取first元素的剩余延迟时间delay
long delay = first.getDelay(NANOSECONDS);
// 如果delay<= 0 说明可以取出了
if (delay <= 0)
return q.poll();
// 走到这里说明delay > 0
if (nanos <= 0)
// 如果没有等待时间了，直接返回null
return null;
first = null; // don't retain ref while waiting
// 这里的逻辑就是取nanos、delay时间中一个小的时间，进行阻塞
if (nanos < delay || leader != null)
nanos = available.awaitNanos(nanos);
else {
// 如果leader为null，设置一下leader为自己
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// 阻塞等待
long timeLeft = available.awaitNanos(delay);
// 计算剩余阻塞时间
nanos -= delay - timeLeft;
} finally {
// 走到这里说明从阻塞状态中被唤醒了
if (leader == thisThread)
// 如果leader发现是自己，现在自己已经不阻塞了，所以需要设置leader=null
leader = null;
}
}
}
}
} finally {
// 判断一下队列中还有元素，此时可能还有别的线程被阻塞者，唤醒一下
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
这里的poll(long timeout,TimeUnit unit)方法里面的源码流程几乎和take方法是一致的。区别只是在于这里计算剩余阻塞时间不一样而已。
因为这里支持阻塞时间自定义，所以每一次阻塞之前，计算最小阻塞时间是多少。苏醒之后计算一下剩余阻塞时间是多少，有没有超时这些逻辑。
其实内部最核心的机制，也就是阻塞、唤醒的机制还是和之前的take方法一样的，这里就不再画图说明了。
4 小结
好了，到这里DelayQueue延迟队列就看的差不多了，有理解不对的地方欢迎指正哈。
标签：JUC,Java,队列,元素,阻塞,延迟时间,线程,DelayQueue,null From： https://www.cnblogs.com/kukuxjx/p/17300783.html