DelayQueue是一个无界阻塞队列，可以用做延时处理，延时处理就是可以为队列中元素设定一个过期时间，队列中的元素按照自定义的delay时间进行排序，其中的元素只有在时间到期时才能从被提取。
队列中的头节点是最接近过期的元素。
该队列中的元素必须实现Delayed接口，该接口强制实现下列两个方法。
compareTo(Delayed o) ：Delayed接口继承了Comparable接口，因此有了这个方法
getDelay(TimeUnit unit)：这个方法返回到激活日期的剩余时间，时间单位由单位参数指定。Delayed接口的方法
该队列实现了Leader-Follower模式,减少不必要的线程等待.关于Leader-Follower可以去看这篇文章：https://www.cnblogs.com/yeyang/p/12580600.html
1.当存在多个take线程时，同时只生效一个（队列中有多个已经过期的元素除外，因为take源码中，如果元素过期了，是直接删除并返回元素的），即leader线程
2.当leader存在时，其它的take线程均为follower，其等待是通过condition实现的
3.当leader不存在时，当前线程即成为leader，在delay之后，将leader角色释放还原
4.最后如果队列还有内容，且leader空缺，则调用一次condition的signal，唤醒挂起的take线程，其中之一将成为新的leader
DelayQueue主体结构：
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E> {
private final transient ReentrantLock lock = new ReentrantLock();
// 主体队列，用作元素的delay时间排序
private final PriorityQueue<E> q = new PriorityQueue<E>();
// leader线程
private Thread leader = null;
// 条件等待队列
private final Condition available = lock.newCondition();
}
DelayQueue构造函数
public DelayQueue() {}
public DelayQueue(Collection<? extends E> c) {
this.addAll(c);
}
入队API：
public boolean add(E e) {
return offer(e);
}
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 新元素入队
q.offer(e);
// 如果入队之后，队头元素就是刚入队的元素。
// 说明入队之前队列是空的，现在有元素了，唤醒take线程
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
public void put(E e) {
offer(e);
}
出队API：
poll 删除并且返回队列的头部
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 获取队头元素
E first = q.peek();
// 队头是空的，或者队头的过期时间没有到，则返回null。否则出队并删除
if (first == null || first.getDelay(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();
// 队头是空的，take线程条件等待
if (first == null)
available.await();
else {
// 若头节点的过期时间到了，直接删除并返回头节点
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
return q.poll();
first = null; // don't retain ref while waiting 疑问点
// 头节点过期时间没到，且leader线程不是空的，当前take线程等待，让leader执行
if (leader != null)
available.await();
else {
// 头节点过期时间没到，且leader线程是空的，让当前线程成为leader线程
// 并且等待delay时间，delay到了，元素就可以被取了
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay);
} finally {
// delay过去了，如果leader还是当前线程，那么把leader置空，留给下一任leader
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// 最后如果没有leader了，并且队列还不是空的，那么唤醒take线程
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
peek,返回队列的头部，并且不删除元素
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return q.peek();
} finally {
lock.unlock();
}
}
疑问点解答
Q:为什么在take方法中，如果队列中有元素，但是过期时间还没到的时候，要写first = null;这一句？
A:如果不设置first = null，将会引起内存泄露。
线程A到达，队首元素没有到期，设置leader = 线程A，并且执行available.awaitNanos(delay);等待元素过期。
这时线程B来了，因为leader != null，则会available.await();阻塞，线程C、D、E同理。
线程A阻塞完毕了，再次循环，获取列首元素成功，出列。
这个时候列首元素应该会被回收掉，但是问题是它还被线程B、线程C持有着，所以不会回收，如果线程增多，且队首元素无限期的不能回收，就会造成内存泄漏。
网友评论