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>(); //优先级队列

 /* Thread designated to wait for the element at the head of the queue. This variant of the Leader-Follower pattern (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves

  * to minimize unnecessary timed waiting. When a thread becomes the leader, it waits only for the next delay to elapse, but other threads await indefinitely.

  * The leader thread must signal some other thread before returning from take() or poll(...), unless some other thread becomes leader in the interim.

  * Whenever the head of the queue is replaced with an element with an earlier expiration time, the leader field is invalidated by being reset to null,

  * and some waiting thread, but not necessarily the current leader, is signalled. So waiting threads must be prepared to acquire and lose leadership while waiting.

  * 指定用于等待队列头部的元素的线程。这种Leader-Follower模式的变体(http://www.cs.wustl.edu/~schmidt/POSA/POSA2/)可以减少不必要的定时等待。

  * 当一个线程成为leader时，它只等待下一个延迟过期，而其他线程则无限期地等待。leader线程必须在从take()或poll(…)返回之前向其他线程发出信号，除非其他线程在此期间成为leader。

  * 每当队列的头部被具有更早过期时间的元素替换时，leader字段就会通过重置为null而无效，并且会通知等待的线程(不一定是当前的leader)的信号。

  * 因此，等待线程必须准备好在等待时获得和失去领导权。

  */

 private Thread leader = null;

 /* Condition signalled when a newer element becomes available at the head of the queue or a new thread may need to become leader.

  * 当队列头部的新元素可用或新线程可能需要成为leader时发出的条件。

  */

 private final Condition available = lock.newCondition();

 package com.Queue;

 import java.util.concurrent.DelayQueue;

 import java.util.concurrent.Delayed;

 import java.util.concurrent.TimeUnit;

 public class DelayQueueTest {

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

         DelayQueue<DelayTask> dq = new DelayQueue();

         //入队四个元素，注意它们的延迟时间单位不一样。

         dq.offer(new DelayTask(5, TimeUnit.SECONDS));

         dq.offer(new DelayTask(2, TimeUnit.MINUTES));

         dq.offer(new DelayTask(700, TimeUnit.MILLISECONDS));

         dq.offer(new DelayTask(1000, TimeUnit.NANOSECONDS));

         while(dq.size() > 0){

             System.out.println(dq.take());

         }

         /*

         打印顺序：

         DelayTask{delay=1000, unit=NANOSECONDS}

         DelayTask{delay=700000000, unit=MILLISECONDS}

         DelayTask{delay=5000000000, unit=SECONDS}

         DelayTask{delay=120000000000, unit=MINUTES}

         */

     }

 }

 class DelayTask implements Delayed {

     private long delay; //延迟多少纳秒开始执行

     private TimeUnit unit;

     public DelayTask(long delay, TimeUnit unit){

         this.unit = unit;

         this.delay = TimeUnit.NANOSECONDS.convert(delay, unit);//统一转换成纳秒计数

     }

     @Override

     public long getDelay(TimeUnit unit) {//延迟剩余时间，单位unit指定

         return unit.convert(delay - System.currentTimeMillis(), TimeUnit.NANOSECONDS);

     }

     @Override

     public int compareTo(Delayed o) {//基于getDelay实时延迟剩余时间进行比较

         if(this.getDelay(TimeUnit.NANOSECONDS) < o.getDelay(TimeUnit.NANOSECONDS)) //都换算成纳秒计算

             return -1;

         else if(this.getDelay(TimeUnit.NANOSECONDS) > o.getDelay(TimeUnit.NANOSECONDS)) //都换算成纳秒计算

             return 1;

         else

             return 0;

     }

     @Override

     public String toString() {

         return "DelayTask{" +

                 "delay=" + delay +

                 ", unit=" + unit +

                 '}';

     }

 }

 //A ScheduledFuture that is Runnable. Successful execution of the run method causes completion of the Future and allows access to its results.

 //run方法的成功执行将导致Future的完成并允许访问其结果。

 public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {

     //如果此任务是周期性的，则返回true。定期任务可以根据某些计划重新运行。非周期任务只能运行一次。

     boolean isPeriodic();

 }

 //A delayed result-bearing action that can be cancelled.  一种可以取消的延迟产生结果的动作。

 //Usually a scheduled future is the result of scheduling a task with a ScheduledExecutorService.

 //通常，ScheduledFuture是ScheduledExecutorService调度任务的结果。

 public interface ScheduledFuture<V> extends Delayed, Future<V> {

         //继承了Delayed接口

 }

 /**

      * Specialized delay queue. To mesh with TPE declarations, this

      * class must be declared as a BlockingQueue<Runnable> even though

      * it can only hold RunnableScheduledFutures.

      */

     static class DelayedWorkQueue extends AbstractQueue<Runnable>

         implements BlockingQueue<Runnable> {

         /*

          * A DelayedWorkQueue is based on a heap-based data structure

          * like those in DelayQueue and PriorityQueue, except that

          * every ScheduledFutureTask also records its index into the

          * heap array. This eliminates the need to find a task upon

          * cancellation, greatly speeding up removal (down from O(n)

          * to O(log n)), and reducing garbage retention that would

          * otherwise occur by waiting for the element to rise to top

          * before clearing. But because the queue may also hold

          * RunnableScheduledFutures that are not ScheduledFutureTasks,

          * we are not guaranteed to have such indices available, in

          * which case we fall back to linear search. (We expect that

          * most tasks will not be decorated, and that the faster cases

          * will be much more common.)

          *

          * All heap operations must record index changes -- mainly

          * within siftUp and siftDown. Upon removal, a task's

          * heapIndex is set to -1. Note that ScheduledFutureTasks can

          * appear at most once in the queue (this need not be true for

          * other kinds of tasks or work queues), so are uniquely

          * identified by heapIndex.

          */

         private static final int INITIAL_CAPACITY = 16;

         private RunnableScheduledFuture<?>[] queue =

             new RunnableScheduledFuture<?>[INITIAL_CAPACITY];

         private final ReentrantLock lock = new ReentrantLock();

         private int size = 0;

         /**

          * Thread designated to wait for the task at the head of the

          * queue.  This variant of the Leader-Follower pattern

          * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to

          * minimize unnecessary timed waiting.  When a thread becomes

          * the leader, it waits only for the next delay to elapse, but

          * other threads await indefinitely.  The leader thread must

          * signal some other thread before returning from take() or

          * poll(...), unless some other thread becomes leader in the

          * interim.  Whenever the head of the queue is replaced with a

          * task with an earlier expiration time, the leader field is

          * invalidated by being reset to null, and some waiting

          * thread, but not necessarily the current leader, is

          * signalled.  So waiting threads must be prepared to acquire

          * and lose leadership while waiting.

          */

         private Thread leader = null;

         /**

          * Condition signalled when a newer task becomes available at the

          * head of the queue or a new thread may need to become leader.

          */

         private final Condition available = lock.newCondition();

 public boolean offer(Runnable x) {

     if (x == null)

         throw new NullPointerException(); //不允许插入null值

     RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;

     final ReentrantLock lock = this.lock;

     lock.lock();

     try {

         int i = size;

         if (i >= queue.length)

             grow();  //类似优先级队列PriorityBlockingQueue的扩容

         size = i + 1;

         if (i == 0) { //队列为空直接放在第一个位置

             queue[0] = e;

             setIndex(e, 0);//这是线程池定时任务特有的逻辑

         } else {

             siftUp(i, e); //类似优先级队列PriorityBlockingQueue的冒泡方式插入元素

         }

         if (queue[0] == e) { //类似延迟队列DelayQueue的消费线程唤醒与leader剥夺

             leader = null;

             available.signal();

         }

     } finally {

         lock.unlock();

     }

     return true;

 }

 public RunnableScheduledFuture<?> take() throws InterruptedException {

     final ReentrantLock lock = this.lock;

     lock.lockInterruptibly();

     try {

         for (;;) {

             RunnableScheduledFuture<?> first = queue[0];//获取不移除head

             if (first == null) //同DelayQueue一样，队列为空，等待生产者offer完唤醒

                 available.await();

             else {

                 long delay = first.getDelay(NANOSECONDS);

                 if (delay <= 0)  //head元素已经过期

                     return finishPoll(first);

                 first = null; // don't retain ref while waiting

                 if (leader != null) //已经有leader在等待了，无限期等待leader醒来通知

                     available.await();

                 else { // 当前线程成为leader

                     Thread thisThread = Thread.currentThread();

                     leader = thisThread;

                     try {

                         available.awaitNanos(delay); //等待heade元素剩余的延迟时间结束

                     } finally {

                         if (leader == thisThread)

                             leader = null;  //交出leader权限

                     }

                 }

             }

         }

     } finally {

         if (leader == null && queue[0] != null) //队列不为空，并且没有leader

             available.signal(); //唤醒其它可能等待的消费线程

         lock.unlock();

     }

 }

 //类似优先级队列PriorityBlockingQueue的出队逻辑

 private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {

     int s = --size;

     RunnableScheduledFuture<?> x = queue[s];

     queue[s] = null;

     if (s != 0)

         siftDown(0, x); //类似PriorityBlockingQueue的拿最后一个元素从head向下降级来确定位置

     setIndex(f, -1);

     return f;

 }