public interface Delayed extends Comparable<Delayed> {
	//獲取元素剩余到期時(shí)間
    long getDelay(TimeUnit unit);
}

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();
    }
}

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        for (;;) {
        	//從優(yōu)先堆中獲取堆頂元素，即優(yōu)先級(jí)最高，即預(yù)定時(shí)間最近的元素
            E first = q.peek();
            if (first == null)
				//若隊(duì)列中無(wú)元素則直接進(jìn)入條件隊(duì)列等待
                available.await();
            else {
                long delay = first.getDelay(NANOSECONDS);
                //若堆頂元素已經(jīng)到期，則直接將其出隊(duì)返回
                if (delay <= 0)
                    return q.poll();
				//等待期間不持有元素引用，防止該元素被其他線程出隊(duì)消費(fèi)后，仍不能被垃圾回收
                first = null; // don't retain ref while waiting
                if (leader != null)
                	//若已經(jīng)有l(wèi)eader了，則進(jìn)入條件隊(duì)列無(wú)限期等待
                    available.await();
                else {
                	//否則成為leader進(jìn)入條件隊(duì)列超時(shí)等待，到預(yù)期時(shí)間或者有更近時(shí)間元素插入就到同步隊(duì)列競(jìng)爭(zhēng)鎖，再重復(fù)循環(huán)去取堆頂元素
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
    	//取完元素后若leader為null且隊(duì)列中還有元素則從條件等待隊(duì)列通知一個(gè)線程到同步隊(duì)列
    	//為什么存在leader不為null的情況：leader線程從awaitNanos()中結(jié)束后沒(méi)有競(jìng)爭(zhēng)過(guò)新進(jìn)take()的線程，因此繼續(xù)在同步隊(duì)列中被阻塞，因此無(wú)需再?gòu)臈l件等待隊(duì)列中通知線程，直接讓leader線程再去競(jìng)爭(zhēng)鎖，
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();//釋放鎖資源讓同步隊(duì)列中的線程競(jìng)爭(zhēng)鎖
    }
}

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 int size() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return q.size();
    } finally {
        lock.unlock();
    }
}

public class DelayQueueTest {
    //隊(duì)列元素類
    static class DelayTask implements Delayed {
        long exeTime;//預(yù)定執(zhí)行時(shí)間
        public DelayTask(long exeTime) {
            this.exeTime = exeTime;
        }
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.exeTime - System.currentTimeMillis(), unit);
        }
        @Override
        public int compareTo(Delayed o) {
            long delta = getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
            return (int) delta;
        }
    }
    public static void main(String[] args) {
        DelayQueue<DelayTask> delayQueue = new DelayQueue<>();
        for (int i = 1;i <= 5;i++) {
            delayQueue.offer(new DelayTask(System.currentTimeMillis() + new Random().nextInt(10)*1000));
        }
        for (int i = 1;i <= 3;i++) {
            new Thread(() -> {
                try {
                    while (true) {
                        DelayTask task = delayQueue.take();
                        System.out.printf("取出任務(wù)！取出時(shí)間：%s 任務(wù)預(yù)定執(zhí)行時(shí)間：%s%n", hms(System.currentTimeMillis()), hms(task.exeTime));
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
    public static String hms(long milliseconds) {
        return new SimpleDateFormat("HH:mm:ss").format(milliseconds);
    }
}