前言

線程池運行時，會不斷從任務(wù)隊列中獲取任務(wù)，然后執(zhí)行任務(wù)。如果我們想實現(xiàn)延時或者定時執(zhí)行任務(wù)，重要一點就是任務(wù)隊列會根據(jù)任務(wù)延時時間的不同進行排序，延時時間越短地就排在隊列的前面，先被獲取執(zhí)行。

隊列是先進先出的數(shù)據(jù)結(jié)構(gòu)，就是先進入隊列的數(shù)據(jù)，先被獲取。但是有一種特殊的隊列叫做優(yōu)先級隊列，它會對插入的數(shù)據(jù)進行優(yōu)先級排序，保證優(yōu)先級越高的數(shù)據(jù)首先被獲取，與數(shù)據(jù)的插入順序無關(guān)。

實現(xiàn)優(yōu)先級隊列高效常用的一種方式就是使用堆。關(guān)于堆的實現(xiàn)可以查看《堆和二叉堆的實現(xiàn)和特性》

ScheduledThreadPoolExecutor線程池

ScheduledThreadPoolExecutor繼承自ThreadPoolExecutor，所以其內(nèi)部的數(shù)據(jù)結(jié)構(gòu)和ThreadPoolExecutor基本一樣，并在其基礎(chǔ)上增加了按時間調(diào)度執(zhí)行任務(wù)的功能，分為延遲執(zhí)行任務(wù)和周期性執(zhí)行任務(wù)。

ScheduledThreadPoolExecutor的構(gòu)造函數(shù)只能傳3個參數(shù)corePoolSize、ThreadFactory、RejectedExecutionHandler，默認maximumPoolSize為Integer.MAX_VALUE。

工作隊列是高度定制化的延遲阻塞隊列DelayedWorkQueue，其實現(xiàn)原理和DelayQueue基本一樣，核心數(shù)據(jù)結(jié)構(gòu)是二叉最小堆的優(yōu)先隊列，隊列滿時會自動擴容，所以offer操作永遠不會阻塞，maximumPoolSize也就用不上了，所以線程池中永遠會保持至多有corePoolSize個工作線程正在運行。

public ScheduledThreadPoolExecutor(int corePoolSize, 
                                   ThreadFactory threadFactory, 
                                   RejectedExecutionHandler handler) { 
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, 
          new DelayedWorkQueue(), threadFactory, handler); 
} 

 DelayedWorkQueue延遲阻塞隊列  

DelayedWorkQueue 也是一種設(shè)計為定時任務(wù)的延遲隊列，它的實現(xiàn)和DelayQueue一樣，不過是將優(yōu)先級隊列和DelayQueue的實現(xiàn)過程遷移到本身方法體中，從而可以在該過程當中靈活的加入定時任務(wù)特有的方法調(diào)用。

工作原理

DelayedWorkQueue的實現(xiàn)原理中規(guī)中矩，內(nèi)部維護了一個以RunnableScheduledFuture類型數(shù)組實現(xiàn)的最小二叉堆，初始容量是16，使用ReentrantLock和Condition實現(xiàn)生產(chǎn)者和消費者模式。

源碼分析

定義

DelayedWorkQueue 的類繼承關(guān)系如下：

其包含的方法定義如下：

成員屬性

// 初始時，數(shù)組長度大小。 
private static final int INITIAL_CAPACITY = 16;         
// 使用數(shù)組來儲存隊列中的元素。 
private RunnableScheduledFuture<?>[] queue =  new RunnableScheduledFuture<?>[INITIAL_CAPACITY];         
// 使用lock來保證多線程并發(fā)安全問題。 
private final ReentrantLock lock = new ReentrantLock();         
// 隊列中儲存元素的大小 
private int size = 0;         
//特指隊列頭任務(wù)所在線程 
private Thread leader = null;         
// 當隊列頭的任務(wù)延時時間到了，或者有新的任務(wù)變成隊列頭時，用來喚醒等待線程 
private final Condition available = lock.newCondition(); 

 DelayedWorkQueue是用數(shù)組來儲存隊列中的元素，核心數(shù)據(jù)結(jié)構(gòu)是二叉最小堆的優(yōu)先隊列，隊列滿時會自動擴容。

構(gòu)造函數(shù)

DelayedWorkQueue 是 ScheduledThreadPoolExecutor 的靜態(tài)類部類，默認只有一個無參構(gòu)造方法。

static class DelayedWorkQueue extends AbstractQueue<Runnable> 
        implements BlockingQueue<Runnable> { 
 // ... 
} 

 入隊方法

DelayedWorkQueue 提供了 put/add/offer(帶時間) 三個插入元素方法。我們發(fā)現(xiàn)與普通阻塞隊列相比，這三個添加方法都是調(diào)用offer方法。那是因為它沒有隊列已滿的條件，也就是說可以不斷地向DelayedWorkQueue添加元素,當元素個數(shù)超過數(shù)組長度時，會進行數(shù)組擴容。 

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

 offer添加元素

ScheduledThreadPoolExecutor提交任務(wù)時調(diào)用的是DelayedWorkQueue.add，而add、put等一些對外提供的添加元素的方法都調(diào)用了offer。

public boolean offer(Runnable x) {             
    if (x == null)                 
        throw new NullPointerException(); 
    RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;             
    // 使用lock保證并發(fā)操作安全 
    final ReentrantLock lock = this.lock; 
    lock.lock();             
    try {                 
        int i = size;                 
        // 如果要超過數(shù)組長度，就要進行數(shù)組擴容 
        if (i >= queue.length)                     
            // 數(shù)組擴容 
            grow();                 
        // 將隊列中元素個數(shù)加一 
        size = i + 1;                 
        // 如果是第一個元素，那么就不需要排序，直接賦值就行了 
        if (i == 0) { 
         queue[0] = e; 
            setIndex(e, 0); 
        } else {                     
            // 調(diào)用siftUp方法，使插入的元素變得有序。 
            siftUp(i, e); 
        }                 
        // 表示新插入的元素是隊列頭，更換了隊列頭， 
        // 那么就要喚醒正在等待獲取任務(wù)的線程。 
        if (queue[0] == e) { 
         leader = null;                     
            // 喚醒正在等待等待獲取任務(wù)的線程 
            available.signal(); 
        } 
    } finally { 
     lock.unlock(); 
    }             
    return true; 
} 

 其基本流程如下：

1. 其作為生產(chǎn)者的入口，首先獲取鎖。
2. 判斷隊列是否要滿了(size >= queue.length)，滿了就擴容grow()。
3. 隊列未滿，size+1。
4. 判斷添加的元素是否是第一個，是則不需要堆化。
5. 添加的元素不是第一個，則需要堆化siftUp。
6. 如果堆頂元素剛好是此時被添加的元素，則喚醒take線程消費。
7. 最終釋放鎖。

offer基本流程圖如下：

擴容grow()

可以看到，當隊列滿時，不會阻塞等待，而是繼續(xù)擴容。新容量newCapacity在舊容量oldCapacity的基礎(chǔ)上擴容50%(oldCapacity >> 1相當于oldCapacity /2)。最后Arrays.copyOf，先根據(jù)newCapacity創(chuàng)建一個新的空數(shù)組，然后將舊數(shù)組的數(shù)據(jù)復制到新數(shù)組中。

private void grow() {             
    int oldCapacity = queue.length;             
    // 每次擴容增加原來數(shù)組的一半數(shù)量。 
    // grow 50% 
    int newCapacity = oldCapacity + (oldCapacity >> 1);  
    if (newCapacity < 0) // overflow 
     newCapacity = Integer.MAX_VALUE;             
    // 使用Arrays.copyOf來復制一個新數(shù)組 
    queue = Arrays.copyOf(queue, newCapacity); 
} 

 向上堆化siftUp

新添加的元素先會加到堆底，然后一步步和上面的父親節(jié)點比較，若小于父親節(jié)點則和父親節(jié)點互換位置，循環(huán)比較直至大于父親節(jié)點才結(jié)束循環(huán)。通過循環(huán)，來查找元素key應(yīng)該插入在堆二叉樹那個節(jié)點位置，并交互父節(jié)點的位置。

向上堆化siftUp的詳細過程可以查看《堆和二叉堆的實現(xiàn)和特性》

private void siftUp(int k, RunnableScheduledFuture<?> key) {             
    // 當k==0時，就到了堆二叉樹的根節(jié)點了，跳出循環(huán) 
    while (k > 0) {                 
        // 父節(jié)點位置坐標, 相當于(k - 1) / 2 
        int parent = (k - 1) >>> 1;                 
        // 獲取父節(jié)點位置元素 
        RunnableScheduledFuture<?> e = queue[parent];                 
        // 如果key元素大于父節(jié)點位置元素，滿足條件，那么跳出循環(huán) 
        // 因為是從小到大排序的。 
        if (key.compareTo(e) >= 0)                     
            break;                 
        // 否則就將父節(jié)點元素存放到k位置 
        queue[k] = e;                 
        // 這個只有當元素是ScheduledFutureTask對象實例才有用，用來快速取消任務(wù)。 
        setIndex(e, k);                 
        // 重新賦值k，尋找元素key應(yīng)該插入到堆二叉樹的那個節(jié)點 
        k = parent; 
    }             
    // 循環(huán)結(jié)束，k就是元素key應(yīng)該插入的節(jié)點位置 
    queue[k] = key; 
    setIndex(key, k); 
} 

 出隊方法

DelayedWorkQueue 提供了以下幾個出隊方法

• take()，等待獲取隊列頭元素
• poll() ，立即獲取隊列頭元素
• poll(long timeout, TimeUnit unit) ，超時等待獲取隊列頭元素

take消費元素

Worker工作線程啟動后就會循環(huán)消費工作隊列中的元素，因為ScheduledThreadPoolExecutor的keepAliveTime=0，所以消費任務(wù)其只調(diào)用了DelayedWorkQueue.take。take基本流程如下：

• 首先獲取可中斷鎖，判斷堆頂元素是否是空，空的則阻塞等待available.await()。
• 堆頂元素不為空，則獲取其延遲執(zhí)行時間delay，delay <= 0說明到了執(zhí)行時間，出隊列finishPoll。
• delay > 0還沒到執(zhí)行時間，判斷l(xiāng)eader線程是否為空，不為空則說明有其他take線程也在等待，當前take將無限期阻塞等待。
• leader線程為空，當前take線程設(shè)置為leader，并阻塞等待delay時長。
• 當前l(fā)eader線程等待delay時長自動喚醒或者被其他take線程喚醒，則最終將leader設(shè)置為null。
• 再循環(huán)一次判斷delay <= 0出隊列。
• 跳出循環(huán)后判斷l(xiāng)eader為空并且堆頂元素不為空，則喚醒其他take線程，最后是否鎖。

public RunnableScheduledFuture<?> take() throws InterruptedException {             
    final ReentrantLock lock = this.lock; 
    lock.lockInterruptibly();             
    try {                 
        for (;;) { 
         RunnableScheduledFuture<?> first = queue[0];                     
            // 如果沒有任務(wù)，就讓線程在available條件下等待。 
            if (first == null) 
             available.await();                     
            else {                         
                // 獲取任務(wù)的剩余延時時間 
                long delay = first.getDelay(NANOSECONDS);                         
                // 如果延時時間到了，就返回這個任務(wù)，用來執(zhí)行。 
                if (delay <= 0)                             
                    return finishPoll(first);                         
                // 將first設(shè)置為null，當線程等待時，不持有first的引用 
                first = null; // don't retain ref while waiting 
 
                // 如果還是原來那個等待隊列頭任務(wù)的線程， 
                // 說明隊列頭任務(wù)的延時時間還沒有到，繼續(xù)等待。 
                if (leader != null) 
                 available.await();                         
                else {                             
                    // 記錄一下當前等待隊列頭任務(wù)的線程 
                    Thread thisThread = Thread.currentThread(); 
                    leader = thisThread;                             
                    try {                                 
                        // 當任務(wù)的延時時間到了時，能夠自動超時喚醒。 
                        available.awaitNanos(delay); 
                    } finally {                                 
                        if (leader == thisThread) 
                         leader = null; 
                    } 
                } 
            } 
        } 
    } finally {                 
        if (leader == null && queue[0] != null)                    // 喚醒等待任務(wù)的線程 
         available.signal(); 
        ock.unlock(); 
    } 
} 

 take基本流程圖如下： 

 

take線程阻塞等待

可以看出這個生產(chǎn)者take線程會在兩種情況下阻塞等待：

• 堆頂元素為空。
• 堆頂元素的delay > 0 。

finishPoll出隊列

堆頂元素delay<=0，執(zhí)行時間到，出隊列就是一個向下堆化的過程siftDown。

// 移除隊列頭元素 
private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {             
    // 將隊列中元素個數(shù)減一 
    int s = --size;             
    // 獲取隊列末尾元素x 
    RunnableScheduledFuture<?> x = queue[s];             
    // 原隊列末尾元素設(shè)置為null 
    queue[s] = null;             
    if (s != 0)                 
        // 因為移除了隊列頭元素，所以進行重新排序。 
        siftDown(0, x); 
    setIndex(f, -1);             
    return f; 
} 

堆的刪除方法主要分為三步：

1. 先將隊列中元素個數(shù)減一;
2. 將原隊列末尾元素設(shè)置成為隊列頭元素，再將隊列末尾元素設(shè)置為null;
3. 調(diào)用setDown(O,x)方法，保證按照元素的優(yōu)先級排序。

向下堆化siftDown

由于堆頂元素出隊列后，就破壞了堆的結(jié)構(gòu)，需要組織整理下，將堆尾元素移到堆頂，然后向下堆化：

• 從堆頂開始，父親節(jié)點與左右子節(jié)點中較小的孩子節(jié)點比較(左孩子不一定小于右孩子)。
• 父親節(jié)點小于等于較小孩子節(jié)點，則結(jié)束循環(huán)，不需要交換位置。
• 若父親節(jié)點大于較小孩子節(jié)點，則交換位置。
• 繼續(xù)向下循環(huán)判斷父親節(jié)點和孩子節(jié)點的關(guān)系，直到父親節(jié)點小于等于較小孩子節(jié)點才結(jié)束循環(huán)。

向下堆化siftDown的詳細過程可以查看《堆和二叉堆的實現(xiàn)和特性》

private void siftDown(int k, RunnableScheduledFuture<?> key) {      
    // 無符號右移，相當于size/2 
    int half = size >>> 1;             
    // 通過循環(huán)，保證父節(jié)點的值不能大于子節(jié)點。 
    while (k < half) {                 
        // 左子節(jié)點, 相當于 (k * 2) + 1 
        int child = (k << 1) + 1;                 
        // 左子節(jié)點位置元素 
        RunnableScheduledFuture<?> c = queue[child];                 
        // 右子節(jié)點, 相當于 (k * 2) + 2 
        int right = child + 1;                 
        // 如果左子節(jié)點元素值大于右子節(jié)點元素值，那么右子節(jié)點才是較小值的子節(jié)點。 
        // 就要將c與child值重新賦值 
        if (right < size && c.compareTo(queue[right]) > 0) 
         c = queue[child = right];                 
        // 如果父節(jié)點元素值小于較小的子節(jié)點元素值，那么就跳出循環(huán) 
        if (key.compareTo(c) <= 0)                     
            break;                 
        // 否則，父節(jié)點元素就要和子節(jié)點進行交換 
        queue[k] = c; 
        setIndex(c, k); 
        k = child; 
    }             
    queue[k] = key; 
    setIndex(key, k); 
} 

 leader線程

leader線程的設(shè)計，是Leader-Follower模式的變種，旨在于為了不必要的時間等待。當一個take線程變成leader線程時，只需要等待下一次的延遲時間，而不是leader線程的其他take線程則需要等leader線程出隊列了才喚醒其他take線程。

poll()

立即獲取隊列頭元素，當隊列頭任務(wù)是null，或者任務(wù)延時時間沒有到，表示這個任務(wù)還不能返回，因此直接返回null。否則調(diào)用finishPoll方法，移除隊列頭元素并返回。

public RunnableScheduledFuture<?> poll() {             
    final ReentrantLock lock = this.lock; 
    lock.lock();             
    try { 
     RunnableScheduledFuture<?> first = queue[0];                 
        // 隊列頭任務(wù)是null，或者任務(wù)延時時間沒有到，都返回null 
        if (first == null || first.getDelay(NANOSECONDS) > 0)                     
            return null;                 
        else 
         // 移除隊列頭元素 
            return finishPoll(first); 
    } finally { 
     lock.unlock(); 
    } 
} 

 poll(long timeout, TimeUnit unit)

超時等待獲取隊列頭元素，與take方法相比較，就要考慮設(shè)置的超時時間，如果超時時間到了，還沒有獲取到有用任務(wù)，那么就返回null。其他的與take方法中邏輯一樣。

public RunnableScheduledFuture<?> poll(long timeout, TimeUnit unit)             
    throws InterruptedException {             
    long nanos = unit.toNanos(timeout);             
    final ReentrantLock lock = this.lock; 
    lock.lockInterruptibly();             
    try {                 
        for (;;) { 
         RunnableScheduledFuture<?> first = queue[0];                     
            // 如果沒有任務(wù)。 
            if (first == null) {                         
             // 超時時間已到，那么就直接返回null 
                if (nanos <= 0)                             
                    return null;                         
                else 
                 // 否則就讓線程在available條件下等待nanos時間 
                    nanos = available.awaitNanos(nanos); 
            } else {                         
                // 獲取任務(wù)的剩余延時時間 
                long delay = first.getDelay(NANOSECONDS);                         
                // 如果延時時間到了，就返回這個任務(wù)，用來執(zhí)行。 
                if (delay <= 0)                             
                    return finishPoll(first);                         
                // 如果超時時間已到，那么就直接返回null 
                if (nanos <= 0)                             
                    return null;                         
                // 將first設(shè)置為null，當線程等待時，不持有first的引用 
                first = null; // don't retain ref while waiting 
                // 如果超時時間小于任務(wù)的剩余延時時間，那么就有可能獲取不到任務(wù)。 
                // 在這里讓線程等待超時時間nanos 
                if (nanos < delay || leader != null) 
                 nanos = available.awaitNanos(nanos);                         
                else { 
                    Thread thisThread = Thread.currentThread(); 
                    leader = thisThread;                             
                    try {                                 
                        // 當任務(wù)的延時時間到了時，能夠自動超時喚醒。 
                        long timeLeft = available.awaitNanos(delay);                                 
                        // 計算剩余的超時時間 
                        nanos -= delay - timeLeft; 
                    } finally {                                 
                        if (leader == thisThread) 
                         leader = null; 
                    } 
                } 
            } 
        } 
    } finally {                 
        if (leader == null && queue[0] != null)                    // 喚醒等待任務(wù)的線程 
         available.signal(); 
        lock.unlock(); 
    } 
} 

 remove刪除指定元素

刪除指定元素一般用于取消任務(wù)時，任務(wù)還在阻塞隊列中，則需要將其刪除。當刪除的元素不是堆尾元素時，需要做堆化處理。

public boolean remove(Object x) { 
    final ReentrantLock lock = this.lock; 
    lock.lock(); 
    try { 
        int i = indexOf(x); 
        if (i < 0) 
            return false; 
        //維護heapIndex 
        setIndex(queue[i], -1); 
        int s = --size; 
        RunnableScheduledFuture<?> replacement = queue[s]; 
        queue[s] = null; 
        if (s != i) { 
            //刪除的不是堆尾元素，則需要堆化處理 
            //先向下堆化 
            siftDown(i, replacement); 
            if (queue[i] == replacement) 
                //若向下堆化后，i位置的元素還是replacement，說明四無需向下堆化的， 
                //則需要向上堆化 
                siftUp(i, replacement); 
        } 
        return true; 
    } finally { 
        lock.unlock(); 
    } 
} 

 總結(jié)

使用優(yōu)先級隊列DelayedWorkQueue，保證添加到隊列中的任務(wù)，會按照任務(wù)的延時時間進行排序，延時時間少的任務(wù)首先被獲取。

1. DelayedWorkQueue的數(shù)據(jù)結(jié)構(gòu)是基于堆實現(xiàn)的;
2. DelayedWorkQueue采用數(shù)組實現(xiàn)堆，根節(jié)點出隊，用最后葉子節(jié)點替換，然后下推至滿足堆成立條件;最后葉子節(jié)點入隊，然后向上推至滿足堆成立條件;
3. DelayedWorkQueue添加元素滿了之后會自動擴容原來容量的1/2，即永遠不會阻塞，最大擴容可達Integer.MAX_VALUE，所以線程池中至多有corePoolSize個工作線程正在運行;
4. DelayedWorkQueue 消費元素take，在堆頂元素為空和delay >0 時，阻塞等待;
5. DelayedWorkQueue 是一個生產(chǎn)永遠不會阻塞，消費可以阻塞的生產(chǎn)者消費者模式;
6. DelayedWorkQueue 有一個leader線程的變量，是Leader-Follower模式的變種。當一個take線程變成leader線程時，只需要等待下一次的延遲時間，而不是leader線程的其他take線程則需要等leader線程出隊列了才喚醒其他take線程。