一、概述

??上篇文章大致介紹了下Queue相關(guān)的一些接口住练，其中就包括DelayQueue所繼承的BlockingQueue接口，本篇文章將來(lái)學(xué)習(xí)下DelayQueue的相關(guān)實(shí)現(xiàn)愁拭。

二讲逛、簡(jiǎn)介

DelayQueue是一個(gè)無(wú)界的阻塞隊(duì)列，并且是一個(gè)延遲隊(duì)列岭埠，還是一個(gè)優(yōu)先級(jí)隊(duì)列盏混。該隊(duì)列中每個(gè)元素都有一個(gè)過(guò)期時(shí)間，只有在過(guò)期時(shí)間到期的時(shí)候惜论，才可以從中獲取元素许赃；也就是說(shuō)往隊(duì)列中插入數(shù)據(jù)時(shí)(生產(chǎn)者)是不會(huì)被阻塞的，而只有在獲取數(shù)據(jù)的時(shí)候(消費(fèi)者)才會(huì)被阻塞来涨。

隊(duì)列的頭部是過(guò)期時(shí)間到期后保存時(shí)間最長(zhǎng)的 Delayed 元素图焰，如果沒(méi)有過(guò)期時(shí)間到期的元素，則隊(duì)列沒(méi)有頭部蹦掐，并且poll將返回null技羔。而當(dāng)一個(gè)元素的 getDelay(TimeUnit.NANOSECONDS) 方法返回一個(gè)小于或等于零的值時(shí)，則說(shuō)明過(guò)期時(shí)間到了卧抗，poll就可以移除這個(gè)元素了藤滥，還有此隊(duì)列不允許使用 null 元素。

三社裆、源碼

接下來(lái)我們就來(lái)簡(jiǎn)單看下該類的源碼拙绊，首先來(lái)看下繼承結(jié)構(gòu)和構(gòu)造方法。

1. 繼承結(jié)構(gòu)及構(gòu)造方法

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
    implements BlockingQueue<E> {

可以看到，DelayQueue實(shí)現(xiàn)了BlockingQueue接口标沪，表明該隊(duì)列是個(gè)阻塞隊(duì)列榄攀；并且繼承自AbstractQueue，AbstractQueue前面我們簡(jiǎn)單說(shuō)過(guò)金句，提供了隊(duì)列的一些接口的常規(guī)實(shí)現(xiàn)檩赢，繼承該類可以最小化實(shí)現(xiàn)該接口所需的工作量。不過(guò)可能需要注意的是违寞，該類沒(méi)有實(shí)現(xiàn)序列化接口贞瞒。

public interface Delayed extends Comparable<Delayed> {
    // 返回0或者負(fù)數(shù)，表示元素已經(jīng)過(guò)期
    long getDelay(TimeUnit unit);
}

而DelayQueue的對(duì)象繼承自Delayed接口趁曼，Delayed接口是用來(lái)標(biāo)記那些應(yīng)該在給定延遲時(shí)間之后執(zhí)行的對(duì)象军浆。該對(duì)象只有一個(gè)方法getDelay，返回該任務(wù)剩余的過(guò)期時(shí)間（需要給定時(shí)間單位）挡闰；由于該對(duì)象擴(kuò)展自Comparable乒融，所以放入該隊(duì)列的元素要實(shí)現(xiàn)對(duì)應(yīng)的compareTo方法，DelayQueue會(huì)通過(guò)這個(gè)對(duì)隊(duì)列里的元素來(lái)進(jìn)行排序尿这。

public DelayQueue() {}
public DelayQueue(Collection<? extends E> c) {
    this.addAll(c);
}

構(gòu)造方法的話簇抵，就比較常規(guī)，提供了兩個(gè)構(gòu)造方法射众，不多說(shuō)了碟摆。

2. 屬性

DelayQueue中包含了以下屬性：

private final transient ReentrantLock lock = new ReentrantLock();
private final PriorityQueue<E> q = new PriorityQueue<E>();
private Thread leader = null;
private final Condition available = lock.newCondition();

這四個(gè)屬性都很重要：

• 可重入鎖ReentrantLock用于多線程下的加鎖操作；
• 優(yōu)先級(jí)隊(duì)列PriorityQueue叨橱，將會(huì)根據(jù)Delay對(duì)象中的時(shí)間排序典蜕；
• Condition對(duì)象，用于線程阻塞和通知的關(guān)鍵對(duì)象罗洗；當(dāng)隊(duì)首有新元素可用或者有新線程成為 leader 時(shí)會(huì)觸發(fā)Condition條件愉舔；
• 線程leader（也就是Leader-Follower 模型中的 leader），用于優(yōu)化阻塞通知的線程對(duì)象伙菜；leader線程用于減少線程間獲取數(shù)據(jù)競(jìng)爭(zhēng)用的轩缤，如果leader不為空，說(shuō)明已經(jīng)有線程在獲取數(shù)據(jù)了贩绕，然后當(dāng)前線程進(jìn)入等待狀態(tài)即可火的；

從這里我們大概就可以看出DelayQueue執(zhí)行的流程了：

DelayQueue內(nèi)部維護(hù)了一個(gè)PriorityQueue，也就是優(yōu)先級(jí)隊(duì)列淑倾；在每次往優(yōu)先級(jí)隊(duì)列中添加元素時(shí)會(huì)以元素的過(guò)期時(shí)間作為排序條件馏鹤，最先過(guò)期的元素放在優(yōu)先級(jí)最高的位置，也就是隊(duì)頭的位置娇哆。所以也可以這么說(shuō)湃累，DelayQueue是一個(gè)使用優(yōu)先隊(duì)列實(shí)現(xiàn)的無(wú)界的阻塞隊(duì)列勃救，優(yōu)先級(jí)隊(duì)列的比較基準(zhǔn)是時(shí)間。

3. 方法

在來(lái)學(xué)習(xí)方法的源碼之前治力，我們先簡(jiǎn)單來(lái)看下Condation對(duì)象蒙秒，這里只簡(jiǎn)單介紹下，后續(xù)會(huì)再來(lái)學(xué)習(xí)這個(gè)接口宵统。

在我們學(xué)習(xí)Synchronized的時(shí)候税肪，我們知道，Synchronized與wait()和nitofy()/notifyAll()方法組合可以實(shí)現(xiàn)線程間的通信榜田，也即是我們所說(shuō)的等待/通知模型；而ReentrantLock同樣也可以實(shí)現(xiàn)該功能锻梳，但需要借助Condition來(lái)實(shí)現(xiàn)箭券，通過(guò)Condiiton接口的await()和signal()/signalAll()來(lái)實(shí)現(xiàn)相同的功能。

接下來(lái)我們就來(lái)看一下各個(gè)方法的實(shí)現(xiàn)疑枯。

3.1 add/put/offer方法

由于add/put(入隊(duì))方法的實(shí)現(xiàn)內(nèi)部是調(diào)用offer方法來(lái)實(shí)現(xiàn)的辩块，并且由于DelayQueue是無(wú)界隊(duì)列，所以offer的超時(shí)方法也是調(diào)用offer方法來(lái)實(shí)現(xiàn)的荆永，所以我們來(lái)看下offer方法的源碼：

public boolean add(E e) {
    return offer(e);
}
public void put(E e) {
    offer(e);
}
public boolean offer(E e, long timeout, TimeUnit unit) {
    return offer(e);
}


public boolean offer(E e) {
    // 獲取可重入鎖
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 往優(yōu)先級(jí)隊(duì)列中添加元素
        q.offer(e);
        // 如果隊(duì)列的頭部元素(優(yōu)先級(jí)最高的元素)等于添加的元素
        if (q.peek() == e) {
            // 將線程leader設(shè)置為null
            leader = null;
            // 喚醒一個(gè)等待的線程
            available.signal();
        }
        // 無(wú)界隊(duì)列都返回true
        return true;
    } finally {
        lock.unlock();
    }
}

首先獲取鎖后废亭，將元素添加到優(yōu)先級(jí)隊(duì)列，如果添加的元素等于隊(duì)列的頭部元素具钥，說(shuō)明當(dāng)前添加的元素優(yōu)先級(jí)最高豆村，也就是即將過(guò)期的，這時(shí)候?qū)拘補(bǔ)valiable條件的線程骂删，通知他們隊(duì)列中有元素了掌动。

3.2 poll方法

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 獲取隊(duì)頭元素
        E first = q.peek();
        // 判斷隊(duì)頭元素是否為空，或者不為空的情況下隊(duì)頭元素是否到期
        if (first == null || first.getDelay(NANOSECONDS) > 0)
            return null;
        else
            return q.poll();
    } finally {
        lock.unlock();
    }
}

poll方法表示返回并刪除隊(duì)列的頭部元素(出隊(duì))宁玫，該方法的流程如下：

• 首先獲取對(duì)象鎖粗恢；
• 其次判斷隊(duì)列是否為空(如果隊(duì)列為空，peek方法會(huì)返回null)欧瘪，隊(duì)列為空眷射，返回null；
• 如果隊(duì)列不為空佛掖，則判斷隊(duì)頭元素是否過(guò)期妖碉，如果對(duì)頭元素沒(méi)有過(guò)期，返回null苦囱；
• 其他情況的話嗅绸，返回并刪除隊(duì)列的頭部元素；

然后撕彤，我們?cè)賮?lái)看下poll重載的另一個(gè)超時(shí)方法：

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    // 首先鱼鸠，將超時(shí)等待時(shí)間轉(zhuǎn)換為納秒單位
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    // 獲取鎖猛拴，如果沒(méi)有獲取到，線程進(jìn)入等待狀態(tài)蚀狰，但該線程能夠響應(yīng)中斷愉昆，即中斷線程的等待狀態(tài)
    lock.lockInterruptibly();
    try {
        // 無(wú)限循環(huán)操作
        for (;;) {
            E first = q.peek();
            // 隊(duì)列是否為空
            if (first == null) {
                // 如果隊(duì)列是空的，并且超時(shí)時(shí)間不大于0麻蹋，直接返回
                if (nanos <= 0)
                    return null;
                else
                    // 如果隊(duì)列為空跛溉，并且超時(shí)時(shí)間大于0，進(jìn)入等待狀態(tài)
                    nanos = available.awaitNanos(nanos);
            } else {
                // 如果隊(duì)列不為空扮授，獲取隊(duì)頭元素剩余的過(guò)期時(shí)間
                long delay = first.getDelay(NANOSECONDS);
                // 隊(duì)頭元素過(guò)期了芳室，出隊(duì)
                if (delay <= 0)
                    return q.poll();
                // 隊(duì)頭元素沒(méi)有過(guò)期，但超時(shí)時(shí)間不大于0刹勃，返回null
                if (nanos <= 0)
                    return null;
                first = null; // don't retain ref while waiting
                // 如果超時(shí)等待時(shí)間 < 元素過(guò)期時(shí)間 或者有其他的線程在獲取數(shù)據(jù)
                if (nanos < delay || leader != null)
                    // 進(jìn)入等待
                    nanos = available.awaitNanos(nanos);
                else {
                    // 這時(shí)候 超時(shí)等待時(shí)間 >= 元素過(guò)期時(shí)間堪侯，并且沒(méi)有其他線程獲取數(shù)據(jù)
                    // 那么把當(dāng)前線程作為leader，表示該leader線程最早開始等待獲取元素
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        // 等待元素過(guò)期荔仁，這時(shí)候會(huì)釋放鎖伍宦；等過(guò)期后再重新獲取鎖
                        long timeLeft = available.awaitNanos(delay);
                        // 重新計(jì)算最新的超時(shí)時(shí)間
                        nanos -= delay - timeLeft;
                    } finally {
                        // 如果當(dāng)前線程仍然是leader線程，操作完成乏梁，清除掉leader
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        // 喚醒對(duì)應(yīng)的線程
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();
    }
}

該方法有點(diǎn)長(zhǎng)次洼，我們來(lái)簡(jiǎn)單分析下：

• 先判斷隊(duì)列是否為空，如果為空遇骑，再判斷超時(shí)時(shí)間是否大于0卖毁，如果不大于0，直接返回null;如果超時(shí)時(shí)間大于0质蕉，進(jìn)入等待狀態(tài)势篡；
• 如果隊(duì)列不為空，判斷元素是否過(guò)期模暗，過(guò)期出隊(duì)禁悠；如果元素沒(méi)有過(guò)期，但超時(shí)時(shí)間不大于0兑宇，直接返回null碍侦；
• 如果隊(duì)列不為空，但不滿足上述條件隶糕，則判斷 超時(shí)等待時(shí)間是否小于元素過(guò)期時(shí)間瓷产，如果小于或者有其他線程再獲取數(shù)據(jù)，進(jìn)入等待狀態(tài)枚驻；
• 如果超時(shí)等待時(shí)間大于元素過(guò)期時(shí)間濒旦，并且沒(méi)有其他線程獲取數(shù)據(jù)，那么把當(dāng)前線程設(shè)為leader線程再登，并等待元素過(guò)期尔邓；
• 最后操作完成之后晾剖，把leader線程設(shè)置為null，讓其他線程接著進(jìn)行操作梯嗽；

這里要知道齿尽，Condition在阻塞時(shí)會(huì)釋放鎖，在被喚醒時(shí)會(huì)再次獲取鎖灯节，當(dāng)釋放鎖后循头，其他線程就會(huì)參與競(jìng)爭(zhēng)，這樣某個(gè)線程就會(huì)成為leader線程了炎疆。

3.3 take方法

take方法同樣用于出隊(duì)卡骂，不過(guò)該方法是阻塞方法，來(lái)簡(jiǎn)單看一下：

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        for (;;) {
            E first = q.peek();
            // 隊(duì)列是否為空形入，為空則進(jìn)入等待莹菱，直到被喚醒
            if (first == null)
                available.await();
            else {
                long delay = first.getDelay(NANOSECONDS);
                // 隊(duì)頭元素是否過(guò)期
                if (delay <= 0)
                    return q.poll();
                first = null; // don't retain ref while waiting
                // 是否有其他元素在獲取元素锅风，如果有粹淋，進(jìn)行等待霜运；
                if (leader != null)
                    available.await();
                else {
                    // 如果沒(méi)有茅坛，將當(dāng)前線程設(shè)置為leader捣作，并等待元素過(guò)期
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        // 喚醒Conditon條件
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();
    }
}

take方法實(shí)現(xiàn)了一種“Leader-Follower”的線程模型态秧。這種模型中所有線程會(huì)有三種身份中的一種：leader惩激、follower夯巷，以及一個(gè)干活中的狀態(tài)：proccesser眶明。它的基本原則就是阴颖，永遠(yuǎn)最多只有一個(gè) leader。而所有 follower 都在等待成為 leader挽放。有關(guān)leader-follower模型，可以參考DelayQueue源碼分析-掘金

只有l(wèi)eader線程會(huì)等待指定時(shí)間后獲得鎖蔓纠，其他線程都會(huì)進(jìn)入無(wú)限期等待辑畦。這也是為什么在DelayQueue中都是使用signal喚醒，而不使用signalAll的原因（只需要一個(gè)線程成為leader線程）腿倚；這個(gè)方法的邏輯和poll方法基本上差不多纯出，這里就不多說(shuō)了。

3.4 peek方法

peek方法比較簡(jiǎn)單敷燎，就是在調(diào)用優(yōu)先級(jí)隊(duì)列PriorityQueue的peek方法時(shí)做了加鎖操作而已：

public E peek() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return q.peek();
    } finally {
        lock.unlock();
    }
}

相似的操作還有size暂筝，clear，toArray,remove等方法硬贯，這里就不多說(shuō)了焕襟。

3.5 drainTo方法

drainTo方法表示刪除隊(duì)列中所有元素，并將這些元素添加到給定集合中饭豹；在DelayQueue中只會(huì)刪除那些過(guò)期的元素鸵赖，我們來(lái)看下：

public int drainTo(Collection<? super E> c) {
    // 傳入的集合不能為空
    if (c == null)
        throw new NullPointerException();
    // 傳入的集合不能時(shí)當(dāng)前對(duì)象
    if (c == this)
        throw new IllegalArgumentException();
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        int n = 0;
        // 遍歷隊(duì)列务漩，獲取隊(duì)列隊(duì)頭元素
        for (E e; (e = peekExpired()) != null;) {
            c.add(e);       // In this order, in case add() throws.
            // 獲取到對(duì)頭元素后，出隊(duì)操作
            q.poll();
            ++n;
        }
        return n;
    } finally {
        lock.unlock();
    }
}

這里循環(huán)的判斷條件卫漫，調(diào)用的是peekExpired()方法菲饼，我們來(lái)看下這個(gè)方法：

private E peekExpired() {
    // assert lock.isHeldByCurrentThread();
    E first = q.peek();
    return (first == null || first.getDelay(NANOSECONDS) > 0) ?
        null : first;
}

這個(gè)方法就比較簡(jiǎn)單了，表示返回隊(duì)頭過(guò)期的元素列赎；如果隊(duì)列為空宏悦，或者元素沒(méi)有過(guò)期，則返回null包吝。

而另一個(gè)重載的方法饼煞，只是在循環(huán)的時(shí)候判斷了下最大數(shù)量限制，這里也不多說(shuō)了：

try {
    int n = 0;
    for (E e; n < maxElements && (e = peekExpired()) != null;) {
        c.add(e);       // In this order, in case add() throws.
        q.poll();
        ++n;
    }
    return n;
} finally {
    lock.unlock();
}

4. 應(yīng)用場(chǎng)景

了解了DelayQueue的實(shí)現(xiàn)原理之后诗越，我們可以很明白的分析出他的應(yīng)用場(chǎng)景：

• 緩存實(shí)現(xiàn)砖瞧，用DelayQueue保存元素在緩存中的有效期，然后用一個(gè)線程循環(huán)查詢delayQueue嚷狞，清除緩存中超時(shí)的數(shù)據(jù)块促；
• 定時(shí)任務(wù)處理，將定時(shí)任務(wù)及和執(zhí)行時(shí)間保存到DelayQueue中床未，然后同樣用一個(gè)線程循環(huán)查找竭翠，獲取到任務(wù)就執(zhí)行；

最后薇搁，我們通過(guò)一個(gè)例子來(lái)加深對(duì)這個(gè)類的理解：
首先斋扰，定義Delayed對(duì)象：

static class DelayObj implements Delayed {
   /** 延遲時(shí)間，毫秒保存 */
   private long delayTime;
   /** 要執(zhí)行的任務(wù) */
   private String data;

   public DelayObj(long delayTime, String data) {
       this.delayTime = delayTime + System.currentTimeMillis();
       this.data = data;
   }
   // get,set方法省略

   // 獲取剩余過(guò)期時(shí)間
   @Override
   public long getDelay(TimeUnit unit) {
       return unit.convert(this.delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
   }

   // 比較方法
   @Override
   public int compareTo(Delayed o) {
       return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
   }

   // 重寫toString方法
   @Override
   public String toString() {
       DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
       String date =  dateTimeFormatter.format(LocalDateTime.ofInstant(Instant.ofEpochMilli(delayTime),ZoneId.systemDefault()));
        return "DelayObj{" +
               "delayTime=" + date +
               ", data='" + data + '\'' +
               '}';
   }
}

然后定義生產(chǎn)及消費(fèi)者啃洋，并進(jìn)行測(cè)試：

public class BlockingQueueTest {
    /** 線程池 */
    private static ExecutorService executorService =  Executors.newFixedThreadPool(3);

    public static void main(String[] args) {
        DelayQueue<DelayObj> delayQueue = new DelayQueue<>();
        //生產(chǎn)者
        producer(delayQueue);
        //消費(fèi)者
        consumer(delayQueue);
    }

    private static void producer(DelayQueue<DelayObj> delayQueue) {
        executorService.submit(() -> {
            for (int i = 1; i <= 10; i++){
                // 延遲時(shí)間分別是1秒到10秒
                DelayObj delayObj = new DelayObj(i * 1000,"test" + i);
                delayQueue.offer(delayObj);
            }
        });
    }
    private static void consumer(DelayQueue<DelayObj> delayQueue) {
        executorService.submit(() -> {
            while (true) {
                DelayObj delayObj = null;
                try {
                    delayObj = delayQueue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 打印隊(duì)列元素
                System.out.println("consumer --> " + delayObj);
            }
        });
    }
}

可以看到传货，我們往隊(duì)列中添加了10個(gè)元素，每個(gè)元素的過(guò)期時(shí)間是1-10秒宏娄，所以打印的時(shí)候问裕，是每隔1秒打印一個(gè)元素：

consumer --> DelayObj{delayTime=2019-02-27 17:05:19, data='test1'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:20, data='test2'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:21, data='test3'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:22, data='test4'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:23, data='test5'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:24, data='test6'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:25, data='test7'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:26, data='test8'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:27, data='test9'}
consumer --> DelayObj{delayTime=2019-02-27 17:05:28, data='test10'}

代碼示例參考自：Java延時(shí)隊(duì)列DelayQueue的使用 - 開源中國(guó)

四、總結(jié)

到這里孵坚，本篇文章就介紹完了僻澎，我們簡(jiǎn)單總結(jié)下：

• DelayQueue是一個(gè)無(wú)界的可延遲的阻塞隊(duì)列，往隊(duì)列中插入數(shù)據(jù)時(shí)不會(huì)阻塞十饥，只有在獲取數(shù)據(jù)到時(shí)候才會(huì)被阻塞窟勃；
• 該隊(duì)列中每個(gè)元素都有一個(gè)過(guò)期時(shí)間，內(nèi)部維護(hù)了一個(gè)優(yōu)先級(jí)隊(duì)列PriorityQueue逗堵，然后通過(guò)元素的過(guò)期時(shí)間進(jìn)行優(yōu)先級(jí)的排序秉氧；
• 該隊(duì)列中的元素要實(shí)現(xiàn)Delayed接口，并實(shí)現(xiàn)它的getDelay方法蜒秤，并且由于該對(duì)象擴(kuò)展自Comparable汁咏，所以放入該隊(duì)列的元素還要實(shí)現(xiàn)對(duì)應(yīng)的compareTo方法亚斋；
• DelayQueue沒(méi)有實(shí)現(xiàn)序列化接口；

本文參考：
《Java并發(fā)編程實(shí)戰(zhàn)》
Java 并發(fā) --- 阻塞隊(duì)列之DelayQueue源碼分析 - csdn.net
DelayQueue--閱讀源碼從jdk開始 - iteye.com