一、概述
上篇文章大致介绍了下Queue相关的一些接口，其中就包括DelayQueue所继承的BlockingQueue接口，本篇文章将来学习下DelayQueue的相关实现。
二、简介
DelayQueue是一个无界的阻塞队列，并且是一个延迟队列，还是一个优先级队列。该队列中每个元素都有一个过期时间，只有在过期时间到期的时候，才可以从中获取元素；也就是说往队列中插入数据时(生产者)是不会被阻塞的，而只有在获取数据的时候(消费者)才会被阻塞。
队列的头部是过期时间到期后保存时间最长的 Delayed 元素，如果没有过期时间到期的元素，则队列没有头部，并且poll将返回null。而当一个元素的 getDelay(TimeUnit.NANOSECONDS) 方法返回一个小于或等于零的值时，则说明过期时间到了，poll就可以移除这个元素了，还有此队列不允许使用 null 元素。
三、源码
接下来我们就来简单看下该类的源码，首先来看下继承结构和构造方法。
1. 继承结构及构造方法
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E> {
可以看到，DelayQueue实现了BlockingQueue接口，表明该队列是个阻塞队列；并且继承自AbstractQueue，AbstractQueue前面我们简单说过，提供了队列的一些接口的常规实现，继承该类可以最小化实现该接口所需的工作量。不过可能需要注意的是，该类没有实现序列化接口。
public interface Delayed extends Comparable<Delayed> {
// 返回0或者负数，表示元素已经过期
long getDelay(TimeUnit unit);
}
而DelayQueue的对象继承自Delayed接口，Delayed接口是用来标记那些应该在给定延迟时间之后执行的对象。该对象只有一个方法getDelay，返回该任务剩余的过期时间（需要给定时间单位）；由于该对象扩展自Comparable，所以放入该队列的元素要实现对应的compareTo方法，DelayQueue会通过这个对队列里的元素来进行排序。
public DelayQueue() {}
public DelayQueue(Collection<? extends E> c) {
this.addAll(c);
}
构造方法的话，就比较常规，提供了两个构造方法，不多说了。
2. 属性
DelayQueue中包含了以下属性：
private final transient ReentrantLock lock = new ReentrantLock();
private final PriorityQueue<E> q = new PriorityQueue<E>();
private Thread leader = null;
private final Condition available = lock.newCondition();
这四个属性都很重要：
- 可重入锁ReentrantLock用于多线程下的加锁操作；
- 优先级队列PriorityQueue，将会根据Delay对象中的时间排序；
- Condition对象，用于线程阻塞和通知的关键对象；当队首有新元素可用或者有新线程成为 leader 时会触发Condition条件；
- 线程leader（也就是Leader-Follower 模型中的 leader），用于优化阻塞通知的线程对象；leader线程用于减少线程间获取数据竞争用的，如果leader不为空，说明已经有线程在获取数据了，然后当前线程进入等待状态即可；
从这里我们大概就可以看出DelayQueue执行的流程了：
DelayQueue内部维护了一个PriorityQueue，也就是优先级队列；在每次往优先级队列中添加元素时会以元素的过期时间作为排序条件，最先过期的元素放在优先级最高的位置，也就是队头的位置。所以也可以这么说，DelayQueue是一个使用优先队列实现的无界的阻塞队列，优先级队列的比较基准是时间。
3. 方法
在来学习方法的源码之前，我们先简单来看下Condation对象，这里只简单介绍下，后续会再来学习这个接口。
在我们学习Synchronized的时候，我们知道，Synchronized与wait()和nitofy()/notifyAll()方法组合可以实现线程间的通信，也即是我们所说的等待/通知模型；而ReentrantLock同样也可以实现该功能，但需要借助Condition来实现，通过Condiiton接口的await()和signal()/signalAll()来实现相同的功能。
接下来我们就来看一下各个方法的实现。
3.1 add/put/offer方法
由于add/put(入队)方法的实现内部是调用offer方法来实现的，并且由于DelayQueue是无界队列，所以offer的超时方法也是调用offer方法来实现的，所以我们来看下offer方法的源码：
public boolean add(E e) {
return offer(e);
}
public void put(E e) {
offer(e);
}
public boolean offer(E e, long timeout, TimeUnit unit) {
return offer(e);
}
public boolean offer(E e) {
// 获取可重入锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 往优先级队列中添加元素
q.offer(e);
// 如果队列的头部元素(优先级最高的元素)等于添加的元素
if (q.peek() == e) {
// 将线程leader设置为null
leader = null;
// 唤醒一个等待的线程
available.signal();
}
// 无界队列都返回true
return true;
} finally {
lock.unlock();
}
}
首先获取锁后，将元素添加到优先级队列，如果添加的元素等于队列的头部元素，说明当前添加的元素优先级最高，也就是即将过期的，这时候将唤醒avaliable条件的线程，通知他们队列中有元素了。
3.2 poll方法
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 获取队头元素
E first = q.peek();
// 判断队头元素是否为空，或者不为空的情况下队头元素是否到期
if (first == null || first.getDelay(NANOSECONDS) > 0)
return null;
else
return q.poll();
} finally {
lock.unlock();
}
}
poll方法表示返回并删除队列的头部元素(出队)，该方法的流程如下：
- 首先获取对象锁；
- 其次判断队列是否为空(如果队列为空，peek方法会返回null)，队列为空，返回null；
- 如果队列不为空，则判断队头元素是否过期，如果对头元素没有过期，返回null；
- 其他情况的话，返回并删除队列的头部元素；
然后，我们再来看下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) {
// 如果队列是空的，并且超时时间不大于0，直接返回
if (nanos <= 0)
return null;
else
// 如果队列为空，并且超时时间大于0，进入等待状态
nanos = available.awaitNanos(nanos);
} else {
// 如果队列不为空，获取队头元素剩余的过期时间
long delay = first.getDelay(NANOSECONDS);
// 队头元素过期了，出队
if (delay <= 0)
return q.poll();
// 队头元素没有过期，但超时时间不大于0，返回null
if (nanos <= 0)
return null;
first = null; // don't retain ref while waiting
// 如果超时等待时间 < 元素过期时间 或者有其他的线程在获取数据
if (nanos < delay || leader != null)
// 进入等待
nanos = available.awaitNanos(nanos);
else {
// 这时候 超时等待时间 >= 元素过期时间，并且没有其他线程获取数据
// 那么把当前线程作为leader，表示该leader线程最早开始等待获取元素
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// 等待元素过期，这时候会释放锁；等过期后再重新获取锁
long timeLeft = available.awaitNanos(delay);
// 重新计算最新的超时时间
nanos -= delay - timeLeft;
} finally {
// 如果当前线程仍然是leader线程，操作完成，清除掉leader
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// 唤醒对应的线程
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
该方法有点长，我们来简单分析下：
- 先判断队列是否为空，如果为空，再判断超时时间是否大于0，如果不大于0，直接返回null;如果超时时间大于0，进入等待状态；
- 如果队列不为空，判断元素是否过期，过期出队；如果元素没有过期，但超时时间不大于0，直接返回null；
- 如果队列不为空，但不满足上述条件，则判断 超时等待时间是否小于元素过期时间，如果小于或者有其他线程再获取数据，进入等待状态；
- 如果超时等待时间大于元素过期时间，并且没有其他线程获取数据，那么把当前线程设为leader线程，并等待元素过期；
- 最后操作完成之后，把leader线程设置为null，让其他线程接着进行操作；
这里要知道，Condition在阻塞时会释放锁，在被唤醒时会再次获取锁，当释放锁后，其他线程就会参与竞争，这样某个线程就会成为leader线程了。
3.3 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 = null; // don't retain ref while waiting
// 是否有其他元素在获取元素，如果有，进行等待；
if (leader != null)
available.await();
else {
// 如果没有，将当前线程设置为leader，并等待元素过期
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// 唤醒Conditon条件
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
take方法实现了一种“Leader-Follower”的线程模型。这种模型中所有线程会有三种身份中的一种：leader、follower，以及一个干活中的状态：proccesser。它的基本原则就是，永远最多只有一个 leader。而所有 follower 都在等待成为 leader。有关leader-follower模型，可以参考DelayQueue源码分析-掘金
只有leader线程会等待指定时间后获得锁，其他线程都会进入无限期等待。这也是为什么在DelayQueue中都是使用signal唤醒，而不使用signalAll的原因（只需要一个线程成为leader线程）；这个方法的逻辑和poll方法基本上差不多，这里就不多说了。
3.4 peek方法
peek方法比较简单，就是在调用优先级队列PriorityQueue的peek方法时做了加锁操作而已：
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return q.peek();
} finally {
lock.unlock();
}
}
相似的操作还有size，clear，toArray,remove等方法，这里就不多说了。
3.5 drainTo方法
drainTo方法表示删除队列中所有元素，并将这些元素添加到给定集合中；在DelayQueue中只会删除那些过期的元素，我们来看下：
public int drainTo(Collection<? super E> c) {
// 传入的集合不能为空
if (c == null)
throw new NullPointerException();
// 传入的集合不能时当前对象
if (c == this)
throw new IllegalArgumentException();
final ReentrantLock lock = this.lock;
lock.lock();
try {
int n = 0;
// 遍历队列，获取队列队头元素
for (E e; (e = peekExpired()) != null;) {
c.add(e); // In this order, in case add() throws.
// 获取到对头元素后，出队操作
q.poll();
++n;
}
return n;
} finally {
lock.unlock();
}
}
这里循环的判断条件，调用的是peekExpired()方法，我们来看下这个方法：
private E peekExpired() {
// assert lock.isHeldByCurrentThread();
E first = q.peek();
return (first == null || first.getDelay(NANOSECONDS) > 0) ?
null : first;
}
这个方法就比较简单了，表示返回队头过期的元素；如果队列为空，或者元素没有过期，则返回null。
而另一个重载的方法，只是在循环的时候判断了下最大数量限制，这里也不多说了：
try {
int n = 0;
for (E e; n < maxElements && (e = peekExpired()) != null;) {
c.add(e); // In this order, in case add() throws.
q.poll();
++n;
}
return n;
} finally {
lock.unlock();
}
4. 应用场景
了解了DelayQueue的实现原理之后，我们可以很明白的分析出他的应用场景：
- 缓存实现，用DelayQueue保存元素在缓存中的有效期，然后用一个线程循环查询delayQueue，清除缓存中超时的数据；
- 定时任务处理，将定时任务及和执行时间保存到DelayQueue中，然后同样用一个线程循环查找，获取到任务就执行；
最后，我们通过一个例子来加深对这个类的理解：
首先，定义Delayed对象：
static class DelayObj implements Delayed {
/** 延迟时间，毫秒保存 */
private long delayTime;
/** 要执行的任务 */
private String data;
public DelayObj(long delayTime, String data) {
this.delayTime = delayTime + System.currentTimeMillis();
this.data = data;
}
// get,set方法省略
// 获取剩余过期时间
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(this.delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
// 比较方法
@Override
public int compareTo(Delayed o) {
return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
}
// 重写toString方法
@Override
public String toString() {
DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
String date = dateTimeFormatter.format(LocalDateTime.ofInstant(Instant.ofEpochMilli(delayTime),ZoneId.systemDefault()));
return "DelayObj{" +
"delayTime=" + date +
", data='" + data + '\'' +
'}';
}
}
然后定义生产及消费者，并进行测试：
public class BlockingQueueTest {
/** 线程池 */
private static ExecutorService executorService = Executors.newFixedThreadPool(3);
public static void main(String[] args) {
DelayQueue<DelayObj> delayQueue = new DelayQueue<>();
//生产者
producer(delayQueue);
//消费者
consumer(delayQueue);
}
private static void producer(DelayQueue<DelayObj> delayQueue) {
executorService.submit(() -> {
for (int i = 1; i <= 10; i++){
// 延迟时间分别是1秒到10秒
DelayObj delayObj = new DelayObj(i * 1000,"test" + i);
delayQueue.offer(delayObj);
}
});
}
private static void consumer(DelayQueue<DelayObj> delayQueue) {
executorService.submit(() -> {
while (true) {
DelayObj delayObj = null;
try {
delayObj = delayQueue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
// 打印队列元素
System.out.println("consumer --> " + delayObj);
}
});
}
}
可以看到，我们往队列中添加了10个元素，每个元素的过期时间是1-10秒，所以打印的时候，是每隔1秒打印一个元素：
consumer --> DelayObj{delayTime=2019-02-27 17:05:19, data='test1'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:20, data='test2'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:21, data='test3'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:22, data='test4'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:23, data='test5'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:24, data='test6'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:25, data='test7'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:26, data='test8'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:27, data='test9'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:28, data='test10'}
代码示例参考自：Java延时队列DelayQueue的使用 - 开源中国
四、总结
到这里，本篇文章就介绍完了，我们简单总结下：
- DelayQueue是一个无界的可延迟的阻塞队列，往队列中插入数据时不会阻塞，只有在获取数据到时候才会被阻塞；
- 该队列中每个元素都有一个过期时间，内部维护了一个优先级队列PriorityQueue，然后通过元素的过期时间进行优先级的排序；
- 该队列中的元素要实现Delayed接口，并实现它的getDelay方法，并且由于该对象扩展自Comparable，所以放入该队列的元素还要实现对应的compareTo方法；
- DelayQueue没有实现序列化接口；
本文参考：
《Java并发编程实战》
Java 并发 --- 阻塞队列之DelayQueue源码分析 - csdn.net
DelayQueue--阅读源码从jdk开始 - iteye.com
网友评论