JUC DelayQueue 分析

心已赠人 2022-05-22 02:20 184阅读 0赞

## 基本介绍 ##

`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中的阻塞队列][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

[Java]: http://www.infoq.com/cn/articles/java-blocking-queue