《java.util.concurrent 包源码阅读》20 DelayQueue

DelayQueue有序存储Delayed类型或者子类型的对象，没当从队列中取走元素时，需要等待延迟耗完才会返回该对象。

所谓Delayed类型，因为需要比较，所以继承了Comparable接口：

public interface Delayed extends Comparable<Delayed> {    long getDelay(TimeUnit unit);}

其实Delayed对象的排序和延迟长短是无关的，因为Comparable的compare方法是用户自己实现的，DelayQueue只是保证返回对象的延迟已经耗尽。

DelayQueue需要排序存储Delayed类型的对象同时具备阻塞功能，但是阻塞的过程伴有延迟等待类型的阻塞，因此不能直接使用BlockingPRiorityQueue来实现，而是用非阻塞的版本的PriorityQueue来实现排序存储。

private final PriorityQueue<E> q = new PriorityQueue<E>();

因此DelayQueue需要自己实现阻塞的功能（需要一个Condition）：

private final Condition available = lock.newCondition();

老规矩还是先来看offer方法：

    public boolean offer(E e) {        final ReentrantLock lock = this.lock;        lock.lock();        try {            q.offer(e);            // 如果原来队列为空，重置leader线程，通知available条件            if (q.peek() == e) {                leader = null;                available.signal();            }            return true;        } finally {            lock.unlock();        }    }

顺便提一下，因为DelayQueue不限制长度，因此添加元素的时候不会因为队列已满产生阻塞，因此带有超时的offer方法的超时设置是不起作用的：

    public boolean offer(E e, long timeout, TimeUnit unit) {        // 和不带timeout的offer方法一样        return offer(e);    }

因为DelayQueue需要自己实现阻塞，因此关注的重点应该是两个带有阻塞的方法：没有超时的take方法和带有超时的poll方法。

普通poll方法很简单，如果延迟时间没有耗尽的话，直接返回null就可以了。

    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和带timeout的poll方法，在看过DelayedWorkQueue之后这部分还是比较好理解的：

    public E take() throws InterruptedException {        final ReentrantLock lock = this.lock;        lock.lockInterruptibly();        try {            for (;;) {                // 如果队列为空，需要等待available条件被通知                E first = q.peek();                if (first == null)                    available.await();                else {                    long delay = first.getDelay(TimeUnit.NANOSECONDS);                    // 如果延迟时间已到，直接返回第一个元素                    if (delay <= 0)                        return q.poll();                    // leader线程存在表示有其他线程在等待，那么当前线程肯定需要等待                    else if (leader != null)                        available.await();                    else {                        Thread thisThread = Thread.currentThread();                        leader = thisThread;                        // 如果没有leader线程，设置当前线程为leader线程                        // 尝试等待直到延迟时间耗尽（可能提前返回，那么下次                        // 循环会继续处理）                        try {                            available.awaitNanos(delay);                        } finally {                            // 如果leader线程还是当前线程，重置它用于下一次循环。                            // 等待available条件时，锁可能被其他线程占用从而导致                            // leader线程被改变，所以要检查                            if (leader == thisThread)                                leader = null;                        }                    }                }            }        } finally {            // 如果没有其他线程在等待，并且队列不为空，通知available条件            if (leader == null && q.peek() != null)                available.signal();            lock.unlock();        }    }

再来看带有timeout的poll方法，和DelayedWorkQueue非常相似：

    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                        // 尝试等待available条件，记录剩余的时间                        nanos = available.awaitNanos(nanos);                } else {                    long delay = first.getDelay(TimeUnit.NANOSECONDS);                    if (delay <= 0)                        return q.poll();                    if (nanos <= 0)                        return null;                    // 当leader线程不为空时（此时delay>=nanos），等待的时间                    // 似乎delay更合理，但是nanos也可以，因为排在当前线程前面的                    // 其他线程返回时会唤醒available条件从而返回，                    // 这里使用nanos和nonas<delay合并更加简单                    if (nanos < delay || leader != null)                        nanos = available.awaitNanos(nanos);                    else {                        Thread thisThread = Thread.currentThread();                        leader = thisThread;                        try {                            long timeLeft = available.awaitNanos(delay);                            // nanos需要更新                            nanos -= delay - timeLeft;                        } finally {                            if (leader == thisThread)                                leader = null;                        }                    }                }            }        } finally {            if (leader == null && q.peek() != null)                available.signal();            lock.unlock();        }    }

前面理解了DelayedWorkQueue再来看DelayQueue就非常容易理解了。