基本介绍

DelayQueue，JUC阻塞队列的一种，依赖ReentrantLock（非公平实现）、Condition以及PriorityQueue来实现阻塞等待获取超时元素。队列中的元素需要超时后才能够被取出，若多个线程尝试获取还没有到超时结束的队列首元素，那么这些线程将会挂起等待(依赖Condition await 以及 超时wait)。DelayQueue是阻塞队列中的一种实现，因此它需要实现BlockingQueue接口，提供阻塞的入队操作和阻塞的出队操作。

注意，DelayQueue是无界的阻塞队列，因此添加操作不会出现阻塞等待。

源码分析

类定义

先看看类定义，以及关键的成员变量:

public class DelayQueue<E extends Delayed> extends AbstractQueue<E> implements BlockingQueue<E> {

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

    // 指向第一个等待获取元素的线程，用于实现Leader/Follower线程模型
    // 对于Leader线程，使用Condition的超时wait
    // 而对于Follower线程，使用Condition的wait
    // 这样做可以尽可能地减少无谓的等待
    // Leader线程在获取元素以后，需要让出leader的地位，唤醒其他Follower，让其竞争成为Leader
    private Thread leader = null;

    // 判断队列是否有元素的Condition
    private final Condition available = lock.newCondition();
}

从上面我们可以看到DelayQueue内部依赖于ReentrantLock、Condition和PriorityQueue。同时注意队列里面的元素必须实现Delayed接口。

下面我们分析入队和出队操作:

入队操作

入队操作包括add、put、offer，实现都在offer方法，超时offer方法最后也调用普通的offer方法

因为DelayQueue是无界队列，没有队列满的情况，offer/put操作没有可能出现阻塞等待，因此超时版的实现就直接调用了普通的offer方法

public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 元素放入优先队列
        q.offer(e);
        // 判断当前优先队列首元素是否为当前添加的元素，
        // 如果是， 需要重置leader，唤醒等待take的线程
        if (q.peek() == e) {
            leader = null;
            available.signal(); // 这里会唤醒第一个等待的线程
        }
        return true;
    } finally {
        lock.unlock();
    }
}

出队操作

出队操作包括非阻塞地出队和阻塞地出队，其中take以及超时的poll会阻塞，而普通的poll不会阻塞。

// 非阻塞出队，获取不到返回null。获取不到包括两种情况:
// 1. 队列为空
// 2. 首元素还没到超时时间
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();
    }
}

// 阻塞的出队
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
                // 有leader线程在等待，当前线程为Follower线程，需阻塞等待
                if (leader != null)
                    available.await();
                // 没有Leader线程，当前Follower线程变成Leader，超时等待
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        // 没有leader线程在等待且队列有元素，则唤醒第一个等待线程
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();
    }
}

阻塞出队还包括超时阻塞出队

// 超时阻塞出队
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
                    nanos = available.awaitNanos(nanos); // 队列无元素，也超时等待
            } else {
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0)
                    return q.poll();
                if (nanos <= 0)
                    return null;
                first = null; // don't retain ref while waiting
                // 当指定的超时时间小于首元素的超时时间 or 存在leader线程，当前线程按指定的超时时间阻塞
                if (nanos < delay || leader != null)
                    nanos = available.awaitNanos(nanos);
                else {
                    // 指定超时时间大于等于首元素超时时间 && 不存在leader线程的情况下
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        long timeLeft = available.awaitNanos(delay);
                        nanos -= delay - timeLeft;
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();
    }
}

使用场景

可以参考聊聊并发（七）——Java中的阻塞队列中的两种使用场景：

• 缓存系统
• 任务调度系统

对于任务调度系统，可以参考JUC中ScheduledThreadPoolExecutor，里面的内部类DelayedWorkQueue，实现的功能也是延迟获取超时的元素，只是比DelayedQueue的功能更加强大也更加复杂

使用示例

首先，队列元素需要实现Delayed接口

public class DelayItem implements Delayed {

    long time;
    String name;

    public DelayItem(long time, String name) {
        this.time = time;
        this.name = name;
    }

    public String toString() {
        return "time : " + time + ", name : " + name;
    }

    @Override
    public int compareTo(Delayed o) {
        if (o == this) 
            return 0;
        long diff = this.getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
        if (diff > 0) {
            return 1;
        } else if (diff < 0) {
            return -1;
        }
        return 0;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(time - System.nanoTime(), TimeUnit.NANOSECONDS);
    }

}

上面是DelayQueue将要存储的元素，简单地，我们只定义了time变量，用于实现getDelay方法和compareTo方法。对于getDelay方法，返回<=0的值，代表的是此元素已经到了超时时间，可以从queue中被取出。而对于compareTo方法，是因为queue里面用了优先队列，队列元素的入队需要根据compareTo方法来决定此元素在优先队列中的位置（有可能后面添加的元素成为新的首元素）。

注意compareTo中小的元素位于队列前面，而越大的元素越往队列后面。

最后，使用示例会往DelayQueue压入两个元素，然后开启两个线程去取元素:

public class DelayQueueTest {

    public static void main(String[] args) throws InterruptedException {

        final BlockingQueue<DelayItem> queue = new DelayQueue<DelayItem>();

        queue.add(new DelayItem(System.nanoTime() + TimeUnit.SECONDS.toNanos(1), "item1"));
        queue.add(new DelayItem(System.nanoTime() + TimeUnit.SECONDS.toNanos(2), "item2"));

        Runnable r = new Runnable() {

            @Override
            public void run() {
                try {
                    DelayItem item = queue.take();
                    System.out.println(item);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        Thread t1 = new Thread(r, "thd1");
        Thread t2 = new Thread(r, "thd2");

        t1.start();
        t2.start();

        Thread.sleep(50);

        t1.join();
        t2.join();
    }

}

输出:

time : 196814664056988, name : item1
time : 196815664590997, name : item2