DelayQueue源码与应用场景分析

先来看DelayQueue继承了哪些类和实现了哪些接口：

DelayQueue是来源于java.util.concurrent包下面的一个类，DelayQueue属于集合中的一员，拥有集合的功能(Collection 接口存储一组不唯一，无序的对象,子类会扩展实现自己的方式)，实现了BlockingQueue接口也属于阻塞队列中的一种，（阻塞这个词来自操作系统的线程/进程的状态模型中，阻塞调用是指调用结果返回之前，调用者会进入阻塞状态等待。只有在得到结果之后才会返回）。

然后我们就来看一下DelayQueue的UML图了解这个类的5大成员

2 DelayQueue的内部成员

2.1 成员变量

1）ReentrantLock重入锁,将由最近成功获得锁，并且还没有释放该锁的线程所拥有。

2）PriorityQueue 一个基于优先级的无界优先级队列,用于存放需要延迟执行的元素。

3）Thread 指定线程等待在队列头部的元素,成员变量Thead leader设计出来是为了minimize unnecessary timed waiting（减少不必要的等待时间）， 在DelayQueue中leader表示一个等待从队列中获取消息的线程。

4）Condition 接口提供一个线程挂起执行的能力，直到给定的条件为真。 Condition对象必须绑定到Lock，并使用newCondition()方法获取对象。

2.2 构造器

1）DelayQueue()

没有参数的构造器，仅仅是用来创建DelayQueue对象。

2）DelayQueue(Collection<? extends E> c)

初始化DelayQueue 并且将参数的集合元素通过offer方法添加到DelayQueue的成员变量PriorityQueue中，集合参数不能为空否则会报空指针异常。

2.3 成员方法

1）public boolean add(E e)

public boolean add(E e) {
      return offer(e);
 }
复制代码

重写父类集合的方法 向延迟队列中插入元素，这里的实现方式是调用offer方法插入元素具体offer方法是怎么插入的可以看offer方法的实现。

2）public boolean offer(E e)

public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.offer(e);
            if (q.peek() == e) {
                leader = null;
                available.signal();
            }
            return true;
        } finally {
            lock.unlock();
        }
    }
复制代码

重写父类Queue类的方法

流程如下：

• 执行加锁操作
• 把元素添加到优先级队列
• 查看元素是否在队列头（也就是第一个元素）
• 果是在队列头部则设置leader对象为空，唤醒所有等待的线程
• 然后释放锁返回true
• 返回值为true添加成功，不会返回false
• q.offer内部是优先级队列的offer 将数据存储在优先级队列的成员 变量transient Object[] queue; queue对象中

3）public void put(E e)

public void put(E e) {
        offer(e);
 }
复制代码

• 重写父类BlockingQueue类中的方法 * 向延迟队列中插入元素，这里的实现方式是调用offer方法插入元素具体offer方法是怎么插入的可以看offer方法的实现。

4） public boolean offer(E e, long timeout, TimeUnit unit)

public boolean offer(E e, long timeout, TimeUnit unit) {
        return offer(e);
 }
复制代码

• 重写父类BlockingQueue类中的方法 *向延迟队列中插入元素，这里的实现方式是调用offer方法插入元素具体offer方法是怎么插入的可以看offer方法的实现。 本方法中timeout参数和unit参数没有使用

4） public E 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();
        }
 }

复制代码

• 重写父接口Queue中的方法 检索并删除此队列的头，如果此队列为空，则返回null

流程如下：

• 加锁
• 查询队列头部元素如果底层的优先级队列中的queue中的头部元素first对象
• 如果取到的元素为空或者调用元素的getDelay方法读取延迟时间大于0还未到延迟等待的结束时间则返回空（这里存放在DelayQueue中的元素必须实现Delayed接口，所有元素都要重写getDelay方法 因为有所有DelayQueue中的元素是有泛型约束 必须要实现Delayed接口的才能成为DelayQueue中的元素的而Delayed接口代码如下

public interface Delayed extends Comparable<Delayed> {

    /**
     *
     *给定时间返回元素
     */
        long getDelay(TimeUnit unit);
     }

复制代码

• 如果执行peek方法可以查询到数据则执行q.poll方法 从优先级队列中的queue数组对象中拉取队列中的头部元素

5） public E take() throws InterruptedException

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;  
                    if (leader != null)
                        available.await();
                    else {
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            available.awaitNanos(delay);
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();
        }
    }
复制代码

• 重写父接口BlockQueue中的方法:检索并删除此队列的头，必要时等待元素变为可用。
• 获取并移除队列头部，在元素变的可用之前一直等待。
• 检索并删除此队列的头，如有必要，请等待直到该队列上具有过期延迟的元素可用。

流程如下：

• 获取全局独占锁，加锁
• 循环起步
• 读取队列头部元素 first
• 如果first为空则执行 available.await(); 进行等待元素存入队列 * 存入元素的offer方法会进行signal唤醒
• 如果first不为空则获取延迟时间delay
• 如果delay小于等于0时间到了则拉取队列头元素同时返回
• delay大于0时first本次拉取的局部变量对象设置为空 防止内存泄漏。然后判断leader对象（从队列中获取消息的线程）不为空则先等待。否则设置leader未当前线程等待延迟时间delay然后等待时间到了之后则会设置leader为空 然后一直进行循环直到上面步骤读取到元素为止读取到则返回
• 方法最终会判断leader(从队列中获取消息的线程)为空并且队列头部元素不为空则发送可用信号让其他等待线程来读取
• 最终释放锁

6）public E poll(long timeout, TimeUnit unit) throws InterruptedException

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; 
                    if (nanos < delay || leader != null)
                        nanos = available.awaitNanos(nanos);
                    else {
                        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();
        }
    }
复制代码

*重写父接口BlockQueue中的方法：检索并删除此队列的头，如果需要，则等待指定的等待时间以使元素变为可用

• 检索并移除此队列的头部，如有必要等待直到到期延迟的元素可以用这个队列，或者在指定的等待时间到期。
• 返回这个队列的头部，或null ，如果指定的等待时间到期延迟元素变得可用之前

流程如下

• 获取全局独占锁，加锁
• 无线循环开始
• 查询队列 头部元素first对象
• 如果队列头部元素为空，同时：如果参数时间小于0 * 则返回空，如果参数时间大于0则等待继续下次循环
• 如果队列头部元素不为空获取元素的getDelay方法中的延迟时间delay变量
• 如果delay变量小于0 则拉取队列头元素同时返回
• 如果delay未小于0 参数等待时间已经小于0 了则返回空
• 设置局部变量first为空
• 如果参数等待时间小于delay元素延迟时间或者leader对象（获取队列元素线程）不为空，则等待参数等待时间
• 如果不满足上面一步则设置leader对象（获取队列元素线程）为当前线程，这个时候delay元素等待时间更小则先等待delay的时间然后参数等待时间减去已经等待的时间则为参数时间未等待的时间更新未等待的时间。最后leader如果为当前线程再将leader设置为空
• 直到时间内读取到元素或者超时未读取到元素循环结束最终如果获取队列元素线程未被其他线程持有并且队列头部存在元素则发送有元素可用信号，最终释放锁

7）public E peek()

public E peek() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.peek();
        } finally {
            lock.unlock();
        }
    }
复制代码

• 重写父接口Queue中的方法检索但不删除此队列的头，如果此队列为空，则返回null 获取，但不移除此队列的头，或者返回null ，如果此队列为空。 * 不同于poll中，如果没有过期元素在队列中是可用的，则此方法返回下一个将到期，如果存在的元素。

流程如下：

• 获取全局独占锁，加锁
• 调用优先级队列的peek方法读取队列头部元素然后进行返回 最终会释放锁

8） public int size()

public int size() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.size();
        } finally {
            lock.unlock();
        }
    }
复制代码

• 获取队列大小
• 重写父接口Collection的方法获取集合中的元素数量

流程如下：

• 加锁
• 读取大小并返回
• 释放锁

9） public int drainTo(Collection<? super E> c)

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();
       }
   }
复制代码

• 重写父接口BlockingQueue阻塞队列中的方法批量获取元素 一次性从BlockingQueue获取队列头连续可用元素（还可以指定获取数据的个数）， 通过该方法，可以提升获取数据效率；不需要多次分批加锁或释放锁。

流程如下：

• 集合参数不能为空
• 集合参数不能为当前延迟队列
• 获取全局独占锁对象然后加锁
• 循环遍历队列中连续的过期可用元素如果元素可用则将元素放入参数集合，如果遇到不可用元素则终止（过期可用元素则看下面peekExpired方法） 循环过程中每遍历一个元素则执行当前内部存储的优先级队列的poll方法移除当前头部元素使下一个元素为头部元素
• 记录读取的元素数量
• 释放锁

private E peekExpired() {
       // assert lock.isHeldByCurrentThread();
       E first = q.peek();
       return (first == null || first.getDelay(NANOSECONDS) > 0) ?  null : first;
   }
复制代码

• 仅当第一个元素过期时返回（过期代表当前元素的getDelay方法返回小于等于0的数
• 仅由drainTo方法调用. 只在持有锁的情况下调用

10 ）public int drainTo(Collection<? super E> c, int maxElements)

public int drainTo(Collection<? super E> c, int maxElements) {
        if (c == null)
            throw new NullPointerException();
        if (c == this)
            throw new IllegalArgumentException();
        if (maxElements <= 0)
            return 0;
        final ReentrantLock lock = this.lock;
        lock.lock();
        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();
        }
    }
复制代码

• 流程与一个参数的drainTo方法基本一致唯一不一样的是循环读取元素时候，循环终止条件变为n < maxElements && (e = peekExpired()) != null 需要同时满足当前已经读取的元素小于最大读取数量和当前队列头部元素可用 重写父接口BlockingQueue阻塞队列中的方法批量获取元素 可以指定获取队列头部可用数据的个数，通过该方法，可以提升获取数据效率；不需要多次分批加锁或释放锁。

11) public void clear()

public void clear() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.clear();
        } finally {
            lock.unlock();
        }
    }
复制代码

• 调用当前内部存储对象优先级队列q将队列中所有元素设置为null 重写父接口Collection的方法清楚集合中的元素

12） public int remainingCapacity()

public int remainingCapacity() {
        return Integer.MAX_VALUE;
}
复制代码

• 总是返回 Integer.MAX_VALUE因为 DelayQueue不受容量限制。 重写父类BlockingQueue

13) public Object[] toArray()

public Object[] toArray() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.toArray();
        } finally {
            lock.unlock();
        }
    }
复制代码

• 重写自Collection父接口的方法将集合转换为数组,内部调用了内部存储对象优先级队列的toArray方法

14） public T[] toArray(T[] a)

public <T> T[] toArray(T[] a) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.toArray(a);
        } finally {
            lock.unlock();
        }
    }
复制代码

• 将集合转换为数组
• 泛型方法
• 返回的数组没有特定顺序

14） public boolean remove(Object o)

public boolean remove(Object o) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.remove(o);
        } finally {
            lock.unlock();
        }
    }
复制代码

• 从该队列中删除指定元素的单个实例（如果存在），无论该实例是否已过期。

15) public Iterator iterator()

public Iterator<E> iterator() {
     return new Itr(toArray());
  }
复制代码

• 继承自Collection的方法获取当前集合的迭代器用来遍历集合用
• 这里将当前集合转换为数组传递给内部迭代器对象然后创建迭代器对象

2.4 内部类

private class Itr implements Iterator<E> {
   final Object[] array; 
     int cursor;       
     int lastRet;        
     Itr(Object[] array) {
         lastRet = -1;
         this.array = array;
     }
     public boolean hasNext() {
         return cursor < array.length;
     }
       @SuppressWarnings("unchecked")
     public E next() {
         if (cursor >= array.length)
             throw new NoSuchElementException();
         lastRet = cursor;
         return (E)array[cursor++];
     }
     public void remove() {
         if (lastRet < 0)
             throw new IllegalStateException();
         removeEQ(array[lastRet]);
         lastRet = -1;
     }
 }
复制代码

• 快照迭代器，用于处理底层q数组的副本

3 DelayQueue的应用场景

1）DelayQueue应用场景，多考生考试问题

模拟一个考试： 考试时间为120分钟 30分钟后才可交卷 考试时间一到，所有未交卷的学生必须交卷 学生都交完卷了考试结束

2）关闭空闲连接。服务器中，有很多客户端的连接，空闲一段时间之后需要关闭之。

3）缓存。缓存中的对象，超过了空闲时间，需要从缓存中移出。

4）任务超时处理。在网络协议滑动窗口请求应答式交互时，处理超时未响应的请求。

5）模拟订单自动取消功能。

6）实现延时消息队列（简易版MQ）。

7）缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询。 DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了。

8）定时任务调度：使用DelayQueue保存当天将会执行的任务和执行时间，一旦从DelayQueue中获取到任务就开始执行，比如TimerQueue就是使用DelayQueue实现的。

等等如果有更多应用场景或者实现可以一块交流。直接在聊天窗口回复消息即可。

5 面试题

String str = "";
System.out.print(str.split(",").length);
复制代码

输出结果为：（不要运行代码，猜你答不对）

• A 0
• B 1
• C 出现异常

答案可以关注微信公众号 拦截器 进行回复。