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

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 {
    //隊列元素類
    static class DelayTask implements Delayed {
        long exeTime;//預定執(zhí)行時間
        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("取出任務！取出時間：%s 任務預定執(zhí)行時間：%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);
    }
}