ScheduledThreadPoolExecutor詳解

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

ScheduledThreadPoolExecutor 的功能與 Timer 類似， 但 ScheduledThreadPoolExecutor 功能更強大、更靈活。

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

ScheduledThreadPoolExecutor 的運行機制

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

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

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

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

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

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

ScheduledThreadPoolExecutor 的實現(xiàn)

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

ScheduledFutureTask 主要包含 3 個成員變量，如下。

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

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

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

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

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

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

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

線程 1 把這個修改 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 時間
                    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 為空，當前線程到 Condition 中等待；否則執(zhí)行下面的 2.2。
• 如果 PriorityQueue 的頭元素的 time 時間比當前時間大，到 Condition 中等待到 time 時間；否則執(zhí)行下面的 2.3。
• 獲取 PriorityQueue 的頭元素（2.3.1）；如果 PriorityQueue 不為空，則喚醒在 Condition 中等待的所有線程（2.3.2）。

3) 釋放 Lock。

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

DelayQueue.offer()方法

// 把 ScheduledFutureTask 放入 DelayQueue 中的過程
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定時任務(wù)詳解的文章就介紹到這了,更多相關(guān)ScheduledThreadPoolExecutor詳解內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論