//可重入同步鎖
private final transient ReentrantLock lock = new ReentrantLock();
//DelayQueue的實(shí)現(xiàn)依賴于PriorityQueue(優(yōu)先隊(duì)列)
private final PriorityQueue<E> q = new PriorityQueue<E>();
//第一個(gè)等待某個(gè)延時(shí)對(duì)象的線程，在延時(shí)對(duì)象還沒有到期時(shí)其他線程看到這個(gè)leader不為null，那么就直接wait
//主要是為了避免大量線程在同一時(shí)間點(diǎn)喚醒，導(dǎo)致大量的競(jìng)爭(zhēng)，反而影響性能
private Thread leader = null;
//條件隊(duì)列，用于wait線程
private final Condition available = lock.newCondition();

    //從上面屬性就可以看出，DelayQueue采用了餓漢模式，調(diào)用構(gòu)造方法即創(chuàng)建了隊(duì)列實(shí)例
    public DelayQueue() {}
    /**
     * 創(chuàng)建一個(gè)DelayQueue，最初包含給定的Collection實(shí)例集合。
     * @param c 最初包含的元素集合
     */
    public DelayQueue(Collection<? extends E> c) {
        this.addAll(c);
    }

public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        //調(diào)用優(yōu)先隊(duì)列
        q.offer(e);
        //檢驗(yàn)元素是否為隊(duì)首，是則設(shè)置 leader 為null, 并喚醒一個(gè)消費(fèi)線程  
        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)先隊(duì)列中獲取第一個(gè)元素，peek方法不會(huì)刪除元素
            E first = q.peek();
            //如果獲取不到數(shù)據(jù)，則調(diào)用available.await()進(jìn)入阻塞狀態(tài)
            if (first == null)
                available.await();
            else {
                //獲取當(dāng)前延時(shí)對(duì)象是否到期
                long delay = first.getDelay(NANOSECONDS);
                //到期那么返回這個(gè)延時(shí)對(duì)象
                if (delay <= 0)
                    return q.poll();
                first = null; // 
                //leader不為空，表明已經(jīng)有其他線程在等待這個(gè)延時(shí)對(duì)象了
                //為什么不available.awaitNanos(delay)呢？這將會(huì)導(dǎo)致大量的線程在同一時(shí)間點(diǎn)被喚醒，然后去競(jìng)爭(zhēng)
                //這個(gè)到期的延時(shí)任務(wù)，影響性能，還不如直接將他們無時(shí)間限制的wait，leader線程或者其他新進(jìn)來的線程獲取到延時(shí)對(duì)象后，去喚醒
                //讓他們?nèi)ジ?jìng)爭(zhēng)下一個(gè)延時(shí)對(duì)象
                if (leader != null)
                    available.await();
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        //指定納秒級(jí)別線程阻塞時(shí)間，當(dāng)前wait住的線程被喚醒后有可能與其他線程競(jìng)爭(zhēng)失敗，就會(huì)進(jìn)入了同步隊(duì)列阻塞，那個(gè)搶到鎖的線程就會(huì)取走這個(gè)延時(shí)對(duì)象
                        available.awaitNanos(delay);
                    } finally {
                        //leader線程被喚醒并獲取到鎖之后會(huì)將leader設(shè)置為空
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        //leader為空并且隊(duì)列不為空，那么喚醒正在等待的線程
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();  //釋放鎖
    }
}

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

public interface Delayed extends Comparable<Delayed> {
    /**
     * 返回與此對(duì)象關(guān)聯(lián)的剩余延遲（給定的時(shí)間單位）。
     * @param unit 時(shí)間單位
     * @返回剩余延遲；零值或負(fù)值表示 延遲已過期
     */
    long getDelay(TimeUnit unit);
}

class MyDelay implements Delayed {
    long delayTime; // 延遲時(shí)間
    long expire; // 過期時(shí)間
    public MyDelay(long delayTime, Thread t) {
        this.delayTime = delayTime;
        // 過期時(shí)間 = 當(dāng)前時(shí)間 + 延遲時(shí)間
        this.expire = System.currentTimeMillis() + delayTime;
    }
    /**
     * 剩余時(shí)間 = 到期時(shí)間 - 當(dāng)前時(shí)間
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }
    /**
     * 優(yōu)先級(jí)規(guī)則：兩個(gè)任務(wù)比較，時(shí)間短的優(yōu)先執(zhí)行
     */
    @Override
    public int compareTo(Delayed o) {
        long f = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
        return (int) f;
    }
}

class MyDelay<T> implements Delayed {
    long delayTime; // 延遲時(shí)間
    long expire; // 過期時(shí)間
    T data;
    public MyDelay(long delayTime, T t) {
        this.delayTime = delayTime;
        // 過期時(shí)間 = 當(dāng)前時(shí)間 + 延遲時(shí)間
        this.expire = System.currentTimeMillis() + delayTime;
        data = t;
    }
    /**
     * 剩余時(shí)間 = 到期時(shí)間 - 當(dāng)前時(shí)間
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }
    /**
     * 優(yōu)先級(jí)規(guī)則：兩個(gè)任務(wù)比較，時(shí)間短的優(yōu)先執(zhí)行
     */
    @Override
    public int compareTo(Delayed o) {
        long f = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
        return (int) f;
    }
    @Override
    public String toString() {
        return "delayTime=" + delayTime +
                ", expire=" + expire +
                ", data=" + data;
    }
}

public class DelayQueueDemo {
    static BlockingQueue<Delayed> queue = new DelayQueue();
    public static void main(String[] args) throws InterruptedException {
        queue.add(new MyDelay(8, "第一次添加任務(wù)"));
        queue.add(new MyDelay(3, "第二次添加任務(wù)"));
        queue.add(new MyDelay(5, "第三次添加任務(wù)"));
        while (!queue.isEmpty()) {
            Delayed delayed = queue.take();
            System.out.println(delayed);
        }
    }
}