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

    //從上面屬性就可以看出，DelayQueue采用了餓漢模式，調(diào)用構(gòu)造方法即創(chuàng)建了隊列實例
    public DelayQueue() {}
    /**
     * 創(chuàng)建一個DelayQueue，最初包含給定的Collection實例集合。
     * @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)先隊列
        q.offer(e);
        //檢驗元素是否為隊首，是則設(shè)置 leader 為null, 并喚醒一個消費線程  
        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)先隊列中獲取第一個元素，peek方法不會刪除元素
            E first = q.peek();
            //如果獲取不到數(shù)據(jù)，則調(diào)用available.await()進(jìn)入阻塞狀態(tài)
            if (first == null)
                available.await();
            else {
                //獲取當(dāng)前延時對象是否到期
                long delay = first.getDelay(NANOSECONDS);
                //到期那么返回這個延時對象
                if (delay <= 0)
                    return q.poll();
                first = null; // 
                //leader不為空，表明已經(jīng)有其他線程在等待這個延時對象了
                //為什么不available.awaitNanos(delay)呢？這將會導(dǎo)致大量的線程在同一時間點被喚醒，然后去競爭
                //這個到期的延時任務(wù)，影響性能，還不如直接將他們無時間限制的wait，leader線程或者其他新進(jìn)來的線程獲取到延時對象后，去喚醒
                //讓他們?nèi)ジ偁幭乱粋€延時對象
                if (leader != null)
                    available.await();
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        //指定納秒級別線程阻塞時間，當(dāng)前wait住的線程被喚醒后有可能與其他線程競爭失敗，就會進(jìn)入了同步隊列阻塞，那個搶到鎖的線程就會取走這個延時對象
                        available.awaitNanos(delay);
                    } finally {
                        //leader線程被喚醒并獲取到鎖之后會將leader設(shè)置為空
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        //leader為空并且隊列不為空，那么喚醒正在等待的線程
        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> {
    /**
     * 返回與此對象關(guān)聯(lián)的剩余延遲（給定的時間單位）。
     * @param unit 時間單位
     * @返回剩余延遲；零值或負(fù)值表示 延遲已過期
     */
    long getDelay(TimeUnit unit);
}

class MyDelay implements Delayed {
    long delayTime; // 延遲時間
    long expire; // 過期時間
    public MyDelay(long delayTime, Thread t) {
        this.delayTime = delayTime;
        // 過期時間 = 當(dāng)前時間 + 延遲時間
        this.expire = System.currentTimeMillis() + delayTime;
    }
    /**
     * 剩余時間 = 到期時間 - 當(dāng)前時間
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }
    /**
     * 優(yōu)先級規(guī)則：兩個任務(wù)比較，時間短的優(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; // 延遲時間
    long expire; // 過期時間
    T data;
    public MyDelay(long delayTime, T t) {
        this.delayTime = delayTime;
        // 過期時間 = 當(dāng)前時間 + 延遲時間
        this.expire = System.currentTimeMillis() + delayTime;
        data = t;
    }
    /**
     * 剩余時間 = 到期時間 - 當(dāng)前時間
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }
    /**
     * 優(yōu)先級規(guī)則：兩個任務(wù)比較，時間短的優(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);
        }
    }
}