public interface Delayed extends Comparable<Delayed> {
    /**
     * 返回關(guān)聯(lián)對(duì)象的剩余延遲時(shí)間（可指定時(shí)間單位）
     */
    long getDelay(TimeUnit unit);
}

public class DelayQueue<E extends Delayed>
        extends AbstractQueue<E>
        implements BlockingQueue<E> {
    /**
     * 可重入鎖，用于保證線程安全
     */
    private final transient ReentrantLock lock = new ReentrantLock();
    /**
     * 優(yōu)先隊(duì)列（容器），實(shí)際存儲(chǔ)元素的地方
     */
    private final PriorityQueue<E> q = new PriorityQueue<E>();
    /**
     * 等待取元素線程的領(lǐng)導(dǎo)(leader)線程，有且僅有一個(gè)leader。
     * 具有最高優(yōu)先級(jí)，第一個(gè)嘗試獲取元素的線程。
     * leader取完元素后，會(huì)喚醒新的等待線程成為新的leader。
     */
    private Thread leader = null;
    /**
     * 觸發(fā)條件，表示是否可以從隊(duì)列中讀取元素.
     * 用于等待（await()）/通知（signal()）其他線程
     */
    private final Condition available = lock.newCondition();
    /**
     * 構(gòu)造函數(shù)
     */
    public DelayQueue() {
    }
    /**
     * 構(gòu)造函數(shù)： 調(diào)用addAll()方法：將集合c 存入隊(duì)列中
     *
     */
    public DelayQueue(Collection<? extends E> c) {
        this.addAll(c);
    }
    /*--------------------------添加元素(非阻塞)-------------------------------*/
    /**
     * 插入新元素.
     * 核心內(nèi)容見(jiàn)：public boolean offer(E e)
     */
    public boolean add(E e) {
        return offer(e);
    }
    /**
     * 插入新元素.
     * 核心內(nèi)容見(jiàn)：public boolean offer(E e)
     */
    public void put(E e) {
        offer(e);
    }
    /**
     * 插入新元素.
     * 核心內(nèi)容見(jiàn)：public boolean offer(E e)
     * @param e       元素
     * @param timeout 此參數(shù)將被忽略，因?yàn)樵摲椒◤牟蛔枞◤U棄）
     * @param unit    此參數(shù)將被忽略，因?yàn)樵摲椒◤牟蛔枞◤U棄）
     * @return {@code true}
     * @throws NullPointerException {@inheritDoc}
     */
    public boolean offer(E e, long timeout, TimeUnit unit) {
        return offer(e);
    }
    /**
     * 插入新元素.(線程安全 lock)
     * 邏輯：
     *  1.入隊(duì)；
     *  2.如果入隊(duì)元素為隊(duì)首元素（原隊(duì)列為空），喚醒一個(gè)等待的線程，通知獲取數(shù)據(jù)。
     *
     * @param e 元素
     * @return {@code true}
     * @throws NullPointerException if the specified element is null
     */
    public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 入隊(duì)
            q.offer(e);
            // 若該元素為隊(duì)列頭部元素（說(shuō)明原隊(duì)列為空），可以喚醒等待的線程取元素?cái)?shù)據(jù)
            if (q.peek() == e) {
                // 如果隊(duì)首元素是剛插入的元素，則設(shè)置leader為null（騰位置）
                leader = null;
                // 喚醒一個(gè)等待的線程
                available.signal();
            }
            return true;
        } finally {
            lock.unlock();
        }
    }
    /*--------------------------取出（返回并刪除）元素-------------------------------*/
    /**
     * 取出延遲到期元素（非阻塞的）.(線程安全 lock)
     * poll() 方法是非阻塞的，即調(diào)用之后無(wú)論元素是否存在/延遲到期都會(huì)立即返回。
     * 邏輯：
     * 1.查詢隊(duì)首元素；
     * 2.元素延遲到期返回，否則返回null
     */
    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 查詢隊(duì)首元素
            E first = q.peek();
            // 隊(duì)首元素為空或者延時(shí)未到期 返回null
            if (first == null || first.getDelay(NANOSECONDS) > 0) {
                return null;
            } else {
                // 如果到期，取出并刪除隊(duì)首元素
                return q.poll();
            }
        } finally {
            lock.unlock();
        }
    }
  /**
     * 取出延遲到期元素（帶有超時(shí)時(shí)間，阻塞）.(線程安全 lock)
     * 如果隊(duì)首元素未到期或者為null，等待：直到隊(duì)首元素延遲到期或者超出指定等待時(shí)間（timeout）
     * 邏輯（無(wú)限循環(huán)等待獲?。?
     * 宗旨：在不超出timeout的時(shí)間內(nèi)，循環(huán)去取出延遲到期的隊(duì)首元素（前提無(wú)其他線程正在取數(shù)--互斥）.
     * 1.查詢隊(duì)首元素；
     *  2.1.隊(duì)列空：等待timeout一段時(shí)間，直到等待超時(shí)（即timeout被重置小于等于0）；
     *  2.2.隊(duì)列不為空：
     *      2.2.1. 隊(duì)首元素延遲到期，取出隊(duì)首元素（poll()）;
     *      2.2.2. 隊(duì)首元素延遲未到期：
     *          2.2.3 等待超時(shí) ，返回null；
     *          2.2.4 等待未超時(shí)，等待時(shí)間<延遲時(shí)間或者有其他線程正在取數(shù)據(jù)，繼續(xù)等待到超時(shí)到期
     *          2.2.5 等待為超時(shí)，等待時(shí)間>=延遲時(shí)間并且無(wú)其他線程正在取數(shù)據(jù)，該線程設(shè)置為leader等待到延遲到期（最后清空l(shuí)eader）
     * 3. 循環(huán)后，如果leader=null（無(wú)正在取數(shù)線程）并且隊(duì)列還有數(shù)據(jù)，喚醒一個(gè)等待線程最終成為leader.
     */
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        // 以可中斷方式獲取鎖
        lock.lockInterruptibly();
        try {
            for (; ; ) {
                // 獲取隊(duì)首元素
                E first = q.peek();
                if (first == null) {
                    // 若隊(duì)首元素為空（即隊(duì)列為空，這時(shí)就需要關(guān)注，當(dāng)前取值請(qǐng)求是否需要阻塞等待
                    // 等待時(shí)間小于等于0 ，不阻塞等待，直接返回null）
                    if (nanos <= 0) {
                        return null;
                    } else {
                        // 等待相應(yīng)的時(shí)間
                        nanos = available.awaitNanos(nanos);
                    }
                } else {
                    // 若隊(duì)列元素非空，獲取隊(duì)首元素剩余延遲時(shí)間
                    long delay = first.getDelay(NANOSECONDS);
                    // 延時(shí)過(guò)期 返回元素
                    if (delay <= 0) {
                        return q.poll();
                    }
                    // 延時(shí)未過(guò)期  等待時(shí)間超時(shí) ，不等待，直接返回null
                    if (nanos <= 0) {
                        return null;
                    }
                    first = null;
                    // 延時(shí)和等待都未到期且等待時(shí)間<延遲時(shí)間 或者 有其他線程在取數(shù)據(jù)，當(dāng)前請(qǐng)求繼續(xù)等待
                    if (nanos < delay || leader != null) {
                        nanos = available.awaitNanos(nanos);
                    } else {
                        // 沒(méi)有其他線程等待，將當(dāng)前線程設(shè)置為 leader，類似于“獨(dú)占”操作
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            // 等待直到延遲到期
                            long timeLeft = available.awaitNanos(delay);
                            // 計(jì)算超時(shí)時(shí)間
                            nanos -= delay - timeLeft;
                        } finally {
                            // 該線程操作完畢，把 leader 置空
                            if (leader == thisThread) {
                                leader = null;
                            }
                        }
                    }
                }
            }
        } finally {
            // 如果leader線程為空 并且  queue非空，則喚醒其他等待線程
            if (leader == null && q.peek() != null) {
                available.signal();
            }
            lock.unlock();
        }
    }
   /**
     * 取出延遲到期元素（無(wú)超時(shí)時(shí)間限制，阻塞）.(線程安全 lock)
     * 邏輯（無(wú)限循環(huán)等待獲?。?
     * 其邏輯參考poll(long timeout, TimeUnit unit).
     * 其區(qū)別在于：不受超時(shí)時(shí)間限制（timeout）
     */
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        // 以可中斷方式獲取鎖
        lock.lockInterruptibly();
        try {
            // 無(wú)限循環(huán)
            for (; ; ) {
                // 獲取隊(duì)首元素
                E first = q.peek();
                if (first == null) {
                    // 若隊(duì)首元素為空（隊(duì)列為空），則等待
                    available.await();
                } else {
                    // 若隊(duì)列元素非空，獲取隊(duì)首元素剩余延遲時(shí)間
                    long delay = first.getDelay(NANOSECONDS);
                    // 延遲到期，獲取隊(duì)首元素
                    if (delay <= 0) {
                        return q.poll();
                    }
                    // 延時(shí)未過(guò)期
                    first = null;
                    // leader 不為空表示有其他線程在讀取數(shù)據(jù)，當(dāng)前線程等待
                    if (leader != null) {
                        available.await();
                    } else {
                        // 沒(méi)有其他線程等待，將當(dāng)前線程設(shè)置為 leader
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            // 等待延遲時(shí)間過(guò)期
                            available.awaitNanos(delay);
                        } finally {
                            if (leader == thisThread) {
                                leader = null;
                            }
                        }
                    }
                }
            }
        } finally {
            // 如果leader線程為空 并且  queue非空，則喚醒其他等待線程
            if (leader == null && q.peek() != null) {
                available.signal();
            }
            lock.unlock();
        }
    }
    /*--------------------------讀取隊(duì)首元素-------------------------------*/
    /**
     * 讀取隊(duì)首元素.（線程安全 lock）
     */
    public E peek() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.peek();
        } finally {
            lock.unlock();
        }
    }
    /*--------------------------讀取隊(duì)列長(zhǎng)度-------------------------------*/
    /**
     * 獲取隊(duì)列數(shù)據(jù)的長(zhǎng)度.（線程安全 lock）
     */
    public int size() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.size();
        } finally {
            lock.unlock();
        }
    }
    /*--------------------------獲取延遲到期元素集合-------------------------------*/
    /**
     * 將隊(duì)列中延遲到期數(shù)據(jù) 收集到集合C中.（線程安全 lock）
     * 
     * @return  返回延遲到期元素?cái)?shù)量
     */
    public int drainTo(Collection<? super E> c) {
        if (c == null) {
            throw new NullPointerException();
        }
        if (c == this) {
            throw new IllegalArgumentException();
        }
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int n = 0;
            //  peekExpired() 判斷隊(duì)首元素是否延遲到期
            for (E e; (e = peekExpired()) != null; ) {
                c.add(e);       
                q.poll();
                ++n;
            }
            return n;
        } finally {
            lock.unlock();
        }
    }
    /**
     * 將隊(duì)列中延遲到期數(shù)據(jù) 收集到集合C中（C集合總數(shù)有限制小于maxElements）.（線程安全 lock）
     * @return  返回延遲到期元素?cái)?shù)量
     */
    public int drainTo(Collection<? super E> c, int maxElements) {
        if (c == null) {
            throw new NullPointerException();
        }
        if (c == this) {
            throw new IllegalArgumentException();
        }
        if (maxElements <= 0) {
            return 0;
        }
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int n = 0;
            // peekExpired() 判斷隊(duì)首元素是否延遲到期。并且到期元素總數(shù)不允許超過(guò)maxElements
            for (E e; n < maxElements && (e = peekExpired()) != null; ) {
                c.add(e);       
                q.poll();
                ++n;
            }
            return n;
        } finally {
            lock.unlock();
        }
    }
    /**
     * 讀取隊(duì)首元素（已延遲到期）.(私有方法)
     */
    private E peekExpired() {
        // 獲取隊(duì)首元素
        E first = q.peek();
        // 隊(duì)首元素存在并且延遲到期，否則返回null
        return (first == null || first.getDelay(NANOSECONDS) > 0) ?
                null : first;
    }
    /*--------------------------刪除元素-------------------------------*/
    /**
     * 清除隊(duì)列中所有元素（線程安全 lock）--暴力清除
     */
    public void clear() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.clear();
        } finally {
            lock.unlock();
        }
    }
    /**
     * 刪除指定元素O.(線程安全 lock)
     */
    public boolean remove(Object o) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.remove(o);
        } finally {
            lock.unlock();
        }
    }
    /**
     * 刪除指定元素O.（這里指的是相同的對(duì)象引用/內(nèi)存地址）(線程安全 lock)
     */
    void removeEQ(Object o) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            for (Iterator<E> it = q.iterator(); it.hasNext(); ) {
                // 使用了對(duì)象引用/內(nèi)存地址相等比較
                if (o == it.next()) {
                    it.remove();
                    break;
                }
            }
        } finally {
            lock.unlock();
        }
    }
    /*--------------------------隊(duì)列轉(zhuǎn)數(shù)組-------------------------------*/
    /**
     * 將隊(duì)列元素都復(fù)制到數(shù)組中（無(wú)序）.(線程安全 lock)
     */
    public Object[] toArray() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.toArray();
        } finally {
            lock.unlock();
        }
    }
    /**
     * 將隊(duì)列元素都復(fù)制到數(shù)組a中（無(wú)序）.
     */
    public <T> T[] toArray(T[] a) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.toArray(a);
        } finally {
            lock.unlock();
        }
    }
    /*--------------------------私有內(nèi)部類--迭代器-------------------------------*/
   /**
     * 返回此隊(duì)列中所有元素（已過(guò)期和未過(guò)期）的迭代器。迭代器不按任何特定順序返回元素。
     */
    public Iterator<E> iterator() {
        return new Itr(toArray());
    }
    /**
     * 快照迭代器，用于處理底層 隊(duì)列/數(shù)組的副本。
     */
    private class Itr implements Iterator<E> {
        final Object[] array; // Array of all elements
        int cursor;           // index of next element to return
        int lastRet;          // index of last element, or -1 if no such
        Itr(Object[] array) {
            lastRet = -1;
            this.array = array;
        }
        public boolean hasNext() {
            return cursor < array.length;
        }
        @SuppressWarnings("unchecked")
        public E next() {
            if (cursor >= array.length)
                throw new NoSuchElementException();
            lastRet = cursor;
            return (E) array[cursor++];
        }
        public void remove() {
            if (lastRet < 0)
                throw new IllegalStateException();
            removeEQ(array[lastRet]);
            lastRet = -1;
        }
    }
}

 @SneakyThrows
    public static void main(String[] args) {
        DelayQueue<TestTask> testTaskDelayQueue = new DelayQueue<>();
        long time = System.currentTimeMillis();
        testTaskDelayQueue.offer(TestTask.builder().name("test_1").endTime(time + 10 * 1000).build());
        testTaskDelayQueue.offer(TestTask.builder().name("test_2").endTime(time + 4 * 1000).build());
        testTaskDelayQueue.offer(TestTask.builder().name("test_3").endTime(time + 16 * 1000).build());
        for(;;){
            System.out.println(testTaskDelayQueue.take());
            TimeUnit.SECONDS.sleep(2);
        }
    }
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    @Builder
    private static class TestTask implements Delayed {
        private String name;
        private Long endTime;
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
        @Override
        public int compareTo(Delayed o) {
            return Long.compare(this.getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
        }
    }