public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
    implements BlockingQueue<E> {
    // 持有內(nèi)部重入鎖。
    private final transient ReentrantLock lock = new ReentrantLock();
    // 優(yōu)先級(jí)隊(duì)列，存放工作任務(wù)。
    private final PriorityQueue<E> q = new PriorityQueue<E>();
    private Thread leader = null;
    // 依賴(lài)于重入鎖的condition。
    private final Condition available = lock.newCondition();
}

/**
 * Inserts the specified element into this delay queue.
 *
 * @param e the element to add
 * @return {@code true} (as specified by {@link Collection#add})
 * @throws NullPointerException if the specified element is null
 */
public boolean add(E e) {
    return offer(e);
}
/**
 * Inserts the specified element into this delay queue. As the queue is
 * unbounded this method will never block.
 *
 * @param e the element to add
 * @throws NullPointerException {@inheritDoc}
 */
public void put(E e) {
    offer(e);
}
/**
 * Inserts the specified element into this delay queue.
 *
 * @param e the element to add
 * @return {@code true}
 * @throws NullPointerException if the specified element is null
 */
public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    //獲取到重入鎖
    lock.lock();
    try {
        q.offer(e);
        //添加成功元素
        if (q.peek() == e) {
            leader = null;
            //將等待隊(duì)列中的頭節(jié)點(diǎn)移動(dòng)到同步隊(duì)列。
            available.signal();
        }
        return true;
    } finally {
        lock.unlock();
    }
}

/**
 * Retrieves and removes the head of this queue, or returns {@code null}
 * if this queue has no elements with an expired delay.
 *
 * @return the head of this queue, or {@code null} if this
 *         queue has no elements with an expired delay
 */
public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        E first = q.peek();
        //如果隊(duì)列為空，或者不為空但是隊(duì)頭元素沒(méi)有過(guò)期則返回null
        if (first == null || first.getDelay(NANOSECONDS) > 0)
            return null;
        else
            return q.poll();
    } finally {
        lock.unlock();
    }
}
/**
 * Retrieves and removes the head of this queue, waiting if necessary
 * until an element with an expired delay is available on this queue.
 *
 * @return the head of this queue
 * @throws InterruptedException {@inheritDoc}
 */
public E take() throws InterruptedException {
    // 獲取鎖。每個(gè)延遲隊(duì)列內(nèi)聚了一個(gè)重入鎖。
    final ReentrantLock lock = this.lock;
    // 獲取可中斷的鎖。
    lock.lockInterruptibly();
    try {
        for (;;) {
            // 嘗試從優(yōu)先級(jí)隊(duì)列中獲取隊(duì)列頭部元素，獲取但不移除
            E first = q.peek();
            if (first == null)
                //無(wú)元素，當(dāng)前線(xiàn)程節(jié)點(diǎn)加入等待隊(duì)列，并阻塞當(dāng)前線(xiàn)程
                available.await();
            else {
                // 通過(guò)延遲任務(wù)的getDelay()方法獲取延遲時(shí)間
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0)
                    //延遲時(shí)間到期，獲取并刪除頭部元素。
                    return q.poll();
                first = null; // don't retain ref while waiting
                if (leader != null)
                    available.await();
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        // 線(xiàn)程節(jié)點(diǎn)進(jìn)入等待隊(duì)列 x 納秒。
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        // 若還存在元素的話(huà)，則將等待隊(duì)列頭節(jié)點(diǎn)中的線(xiàn)程節(jié)點(diǎn)移動(dòng)到同步隊(duì)列中。
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();
    }
}

public class DelayQueueTest {
    public static final int SIZE = 10;
    public static void main(String[] args) {
    	DelayQueueTest test = new DelayQueueTest();
        //初始化線(xiàn)程池
        BlockingQueue<Runnable> arrayBlockingQueue = new ArrayBlockingQueue<>(10);
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor
            (5, 10, 10, TimeUnit.MILLISECONDS,
                arrayBlockingQueue, Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
        DelayQueue<DelayedTask> delayTaskQueue = new DelayQueue<>();
        //模擬SIZE個(gè)延遲任務(wù)
        for (byte i = 0; i < SIZE; i++) {
            Long runAt = System.currentTimeMillis() + 1000 * i;
            String name = "Zhang_" + i;
            byte age = (byte)(10 + i);
            String gender = (i % 2 == 0 ? "male" : "female");
            Student student = new StudentBuilder(name, age, gender).height(150 + i).province("ZheJiang").build();
            delayTaskQueue.put(new DelayedTask<Student>(student, 1, function -> test.print(student), runAt));
        }
        while (true) {
            if (delayTaskQueue.size() == 0) {
                break;
            }
            try {
                //從延遲隊(duì)列中取值,如果沒(méi)有對(duì)象過(guò)期則取到null
                DelayedTask delayedTask = delayTaskQueue.poll();
                if (delayedTask != null) {
                    threadPool.execute(delayedTask);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        threadPool.shutdown();
    }
    public String print(Object object) {
      System.out.println(Thread.currentThread().getName());
        String str = ">>>junit log>>>" + object.getClass().getSimpleName() + ":" + object.toString();
        System.out.println(str);
        return str;
    }
    private static class DelayedTask<T> implements Delayed, Runnable {
        /**
         * 任務(wù)參數(shù)
         */
        private T taskParam;
        /**
         * 任務(wù)類(lèi)型
         */
        private Integer type;
        /**
         * 任務(wù)函數(shù)
         */
        private Function<T, String> function;
        /**
         * 任務(wù)執(zhí)行時(shí)刻
         */
        private Long runAt;
        public T getTaskParam() {
            return taskParam;
        }
        public Integer getType() {
            return type;
        }
        public Function<T, String> getFunction() {
            return function;
        }
        public Long getRunAt() {
            return runAt;
        }
        DelayedTask(T taskParam, Integer type, Function<T, String> function, Long runAt) {
            this.taskParam = taskParam;
            this.type = type;
            this.function = function;
            this.runAt = runAt;
        }
        @Override
        public void run() {
            if (taskParam != null) {
                function.apply(taskParam);
            }
        }
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.runAt - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
        @Override
        public int compareTo(Delayed o) {
            DelayedTask object = (DelayedTask)o;
            return this.runAt.compareTo(object.getRunAt());
        }
    }
}