ScheduledThreadPoolExecutor詳解

ScheduledThreadPoolExecutor 繼承自 ThreadPoolExecutor。它主要用來(lái)在給定的延遲之后運(yùn)行任務(wù)，或者定期執(zhí)行任務(wù)。

ScheduledThreadPoolExecutor 的功能與 Timer 類似， 但 ScheduledThreadPoolExecutor 功能更強(qiáng)大、更靈活。

Timer 對(duì)應(yīng)的是單個(gè)后臺(tái)線程，而 ScheduledThreadPoolExecutor 可以在構(gòu)造函數(shù)中指定多個(gè)對(duì)應(yīng)的后臺(tái)線程數(shù)。

ScheduledThreadPoolExecutor 的運(yùn)行機(jī)制

ScheduledThreadPoolExecutor 的任務(wù)傳遞示意圖

ScheduledThreadPoolExecutor 的執(zhí)行主要分為兩大部分：

1) 當(dāng)調(diào)用 ScheduledThreadPoolExecutor 的 scheduleAtFixedRate()方法或者 scheduleWithFixedDelay()方法時(shí)，會(huì)向 ScheduledThreadPoolExecutor 的 DelayQueue 添加一個(gè)實(shí)現(xiàn)了 RunnableScheduledFutur 接口的 ScheduledFutureTask。

2) 線程池中的線程從 DelayQueue 中獲取 ScheduledFutureTask，然后執(zhí)行任務(wù)。

ScheduledThreadPoolExecutor 為了實(shí)現(xiàn)周期性的執(zhí)行任務(wù)，對(duì) ThreadPoolExecutor 做 了如下的修改：

•  使用 DelayQueue 作為任務(wù)隊(duì)列。
• 獲取任務(wù)的方式不同。
• 執(zhí)行周期任務(wù)后，增加了額外的處理。

ScheduledThreadPoolExecutor 的實(shí)現(xiàn)

ScheduledThreadPoolExecutor 會(huì)把待調(diào)度的任務(wù) （ScheduledFutureTask）放到一個(gè) DelayQueue 中。

ScheduledFutureTask 主要包含 3 個(gè)成員變量，如下。

• long 型成員變量 time，表示這個(gè)任務(wù)將要被執(zhí)行的具體時(shí)間。
• long 型成員變量 sequenceNumber，表示這個(gè)任務(wù)被添加到 ScheduledThreadPoolExecutor 中的序號(hào)。
• long 型成員變量 period，表示任務(wù)執(zhí)行的間隔周期。

DelayQueue 封裝了一個(gè) PriorityQueue，這個(gè) PriorityQueue 會(huì)對(duì)隊(duì)列中的 ScheduledFutureTask 進(jìn)行排序。排序時(shí)，time 小的排在前面（時(shí)間早的任務(wù)將被先執(zhí)行）。

如果兩個(gè) ScheduledFutureTask 的 time 相同，就比較 sequenceNumber，sequenceNumber 小的排在前面（也就是說(shuō)，如果兩個(gè)任務(wù)的執(zhí)行時(shí)間相同，那么先提交的任務(wù)將被先執(zhí)行）。

ScheduledThreadPoolExecutor 中的線程執(zhí)行周期任務(wù)的過(guò)程

線程 1 從 DelayQueue 中獲取已到期的 ScheduledFutureTask（DelayQueue.take()）。

到 期任務(wù)是指 ScheduledFutureTask 的 time 大于等于當(dāng)前時(shí)間。線程 1 執(zhí)行這個(gè) ScheduledFutureTask。

線程 1 修改 ScheduledFutureTask 的 time 變量為下次將要被執(zhí)行的時(shí)間。

線程 1 把這個(gè)修改 time 之后的 ScheduledFutureTask 放回 DelayQueue 中（DelayQueue.add()）。

DelayQueue.take()方法

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly(); // 1  獲取響應(yīng)打斷的鎖
    try {
        for (; ; ) {
            E first = q.peek(); // 獲取ScheduledFutureTask 
            if (first == null) {
                available.await(); // 2.1 到 Condition 中等待
            } else {
                long delay = first.getDelay(TimeUnit.NANOSECONDS);
                if (delay > 0) {
                    // 2.2 到 Condition 中等待到 time 時(shí)間
                    long tl = available.awaitNanos(delay); 
                } else {
                    E x = q.poll(); // 2.3.1 獲取 PriorityQueue 的頭元素
                    assert x != null;
                    if (q.size() != 0)
                        available.signalAll(); // 2.3.2 喚醒在 Condition 中等待的所有線程
                    return x;
                }
            }
        }
    } finally {
        lock.unlock(); // 3 釋放??
    }
}

1) 獲取 Lock。

2) 獲取周期任務(wù)。

• 如果 PriorityQueue 為空，當(dāng)前線程到 Condition 中等待；否則執(zhí)行下面的 2.2。
• 如果 PriorityQueue 的頭元素的 time 時(shí)間比當(dāng)前時(shí)間大，到 Condition 中等待到 time 時(shí)間；否則執(zhí)行下面的 2.3。
• 獲取 PriorityQueue 的頭元素（2.3.1）；如果 PriorityQueue 不為空，則喚醒在 Condition 中等待的所有線程（2.3.2）。

3) 釋放 Lock。

ScheduledThreadPoolExecutor 在一個(gè)循環(huán)中執(zhí)行步驟 2，直到線程從 PriorityQueue 獲 取到一個(gè)元素之后（執(zhí)行 2.3.1 之后），才會(huì)退出無(wú)限循環(huán)（結(jié)束步驟 2）。

DelayQueue.offer()方法

// 把 ScheduledFutureTask 放入 DelayQueue 中的過(guò)程
public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock(); // 1
    try {
        E first = q.peek();
        q.offer(e); // 2.1
        if (first == null || e.compareTo(first) < 0) available.signalAll(); // 2.2
        return true;
    } finally {
        lock.unlock(); // 3
    }
}

1) 獲取 Lock。

2) 添加任務(wù)。

• 向 PriorityQueue 添加任務(wù)。
• 如果在上面 2.1 中添加的任務(wù)是 PriorityQueue 的頭元素，喚醒在 Condition 中等 待的所有線程。

3) 釋放 Lock

到此這篇關(guān)于Java中的ScheduledThreadPoolExecutor定時(shí)任務(wù)詳解的文章就介紹到這了,更多相關(guān)ScheduledThreadPoolExecutor詳解內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論