ScheduledExecutorService 定時任務運行原理（三）

Executor

JUC 包中的 Executor 架構帶來了線程的創建與執行的分離。Executor 的繼承者 ExecutorService 下面衍生出了兩個重要的實現類，他們分別是

• ThreadPoolExecutor 線程池
• ScheduledThreadPoolExecutor 支持周期性任務的線程池

通過 ThreadPoolExecutor 可以實現各式各樣的自定義線程池，而 ScheduledThreadPoolExecutor 類則在自定義線程池的基礎上增加了周期性執行任務的功能。

 

 

使用實例

import java.time.LocalDateTime;

/**
 * 工作任務
 */
public class WorkerThread implements Runnable {

    private String command;

    public WorkerThread(String s) {
        this.command = s;
    }

    @Override
    public void run() {
        
        System.out.println(Thread.currentThread().getName() + " Start. Time = " + LocalDateTime.now());
        processCommand();
        System.out.println(Thread.currentThread().getName() + " End. Time = " + LocalDateTime.now());
    }

    private void processCommand() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString() {
        return this.command;
    }
}

/**
 * 線程池測試類
 *
 * @author lincanhan
 */
public class ScheduledExecutorServiceTest {


    public static void main(String[] args) {
        scheduleWithDelay();
        //scheduleAtRate();
    }


    /**
     * scheduleWithFixedDelay 中的 delayTime
     * 代表每次線程任務執行完畢后，直到下一次開始執行開始之前的時間間隔。
     */public static void scheduleWithDelay() {
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
        WorkerThread workerThread = new WorkerThread("do some thing");
        scheduledExecutorService.scheduleWithFixedDelay(workerThread, 3000, 3000, TimeUnit.MILLISECONDS);
    }

    /**
     * scheduleAtFixedRate 中的 delayTime/period 表示從線程池中首先開始執行的線程算起，假設period為1s，
     * 若線程執行了5s，那么下一個線程在第一個線程運行完后會很快被執行。
     */
    public static void scheduleAtRate() {
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
        WorkerThread workerThread = new WorkerThread("do some thing");
        scheduledExecutorService.scheduleAtFixedRate(workerThread, 3000, 3000, TimeUnit.MILLISECONDS);
    }

}

pool-1-thread-1 Start. Time = 2017-04-06T23:07:36.575
pool-1-thread-1 End. Time = 2017-04-06T23:07:41.579
pool-1-thread-1 Start. Time = 2017-04-06T23:07:44.582
pool-1-thread-1 End. Time = 2017-04-06T23:07:49.586
pool-1-thread-2 Start. Time = 2017-04-06T23:07:52.592
pool-1-thread-2 End. Time = 2017-04-06T23:07:57.597
pool-1-thread-2 Start. Time = 2017-04-06T23:08:00.603
pool-1-thread-2 End. Time = 2017-04-06T23:08:05.608
pool-1-thread-2 Start. Time = 2017-04-06T23:08:08.612
pool-1-thread-2 End. Time = 2017-04-06T23:08:13.616
pool-1-thread-2 Start. Time = 2017-04-06T23:08:16.619

通過使用 ScheduledExecutorService，很方便的實現了每3秒執行一次任務的需求。

下面通過幾個關鍵性的 api 入手，逐步分析 ScheduledExecutorService 是如何實現定時任務的功能的。

newScheduledThreadPool 方法

/**
     * Creates a thread pool that can schedule commands to run after a
     * given delay, or to execute periodically.
     * @param corePoolSize the number of threads to keep in the pool,
     * even if they are idle
     * @return a newly created scheduled thread pool
     * @throws IllegalArgumentException if {@code corePoolSize < 0}
     */
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }


 /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     * @throws IllegalArgumentException   {@inheritDoc}
     */
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (delay <= 0)
            throw new IllegalArgumentException();
            // 將線程任務結合參數，封裝為 DelayedTask
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(-delay));
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

 

將線程任務結合構造參數，封裝為一個 DelayedFutureTask。 如果你還不了解什么是 DelayedTask，請參考我之前的一篇文章： 深入 DelayQueue 內部實現
另外請注意，DelayedFutureTask 是默認實現了 Compare 接口的。

public int compareTo(Delayed other) {
            if (other == this) // compare zero if same object
                return 0;
            if (other instanceof ScheduledFutureTask) {
                ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
                long diff = time - x.time;
                if (diff < 0)
                    return -1;
                else if (diff > 0)
                    return 1;
                else if (sequenceNumber < x.sequenceNumber)
                    return -1;
                else
                    return 1;
            }
            long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
            return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
        }

 

delayedExecute 方法

 private void delayedExecute(RunnableScheduledFuture<?> task) {
        // 判斷線程池是否已經關閉。
        if (isShutdown())
            // 執行拒絕策略。
            reject(task);
        else {
            // 添加封裝好的延遲任務到阻塞隊列中。
            super.getQueue().add(task);
            if (isShutdown() &&
                !canRunInCurrentRunState(task.isPeriodic()) &&
                remove(task))
                task.cancel(false);
            else
                ensurePrestart();
        }
    }

只要線程池正常運行，則將 DelayedTask 添加到 workQueue 中。注意，workQueue 是定義在 ThreadPoolExecutor 當中的用來保存工作任務的阻塞隊列。

ensurePrestart 方法

  /**
     * Same as prestartCoreThread except arranges that at least one
     * thread is started even if corePoolSize is 0.
     */
    void ensurePrestart() {
        int wc = workerCountOf(ctl.get());
        // 當前工作的線程是否超過核心線程數。
        if (wc < corePoolSize)
            // 調用 addWorker 方法，創建新的線程執行任務。
            addWorker(null, true);
        else if (wc == 0)
            addWorker(null, false);
    }

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            // 創建工作線程。
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());

                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        //將工作線程加入到工作者隊列中。工作者隊列是保存工作者線程對象的集合。與工作任務隊列是不一樣的概念。
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                // 啟動創建的工作者線程
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

/**
 * Creates with given first task and thread from ThreadFactory.
 * @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

雖然 addWorker 的代碼稍顯復雜，不過你完全可以無視那一大塊兒 CAS 處理，把注意力放到僅僅與定時任務主要邏輯緊密相關的地方上來。
ensurePrestart 在 addWorker 中創建了工作者線程，添加到工作者隊列以后啟動線程。注意工作者線程不是工作任務線程。工作者線程是線程池啟動的用來執行任務的線程。

Worker Run 方法

/** Delegates main run loop to outer runWorker  */
        public void run() {
            runWorker(this);
        }

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {  
            // 嘗試獲取、或從延遲阻塞隊列中獲取線程任務對象。
            while (task != null || (task = getTask()) != null) {
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                    // 執行任務的 run() 方法。此處直接調用 run 而不調用 start 方法的原因是因為本身已經處在 worker 工作者線程中了。
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

工作者通過循環從工作者任務隊列中反復獲取任務對象，獲取成功后調用對象的 run 方法執行線程任務。當執行結束時，工作者繼續嘗試從延遲阻塞工作者任務隊列中后去新的任務對象進行執行。

無論有幾個工作者線程，無論某個工作者線程處理了幾個任務，任務的間隔總是由定義在每個任務對象中的間隔時間來決定的。

ScheduledFutureTask run 方法

 /**
         * Overrides FutureTask version so as to reset/requeue if periodic.
         */
        public void run() {
            boolean periodic = isPeriodic();
            if (!canRunInCurrentRunState(periodic))
                cancel(false);
                // 若無間隔時間，則直接調用 futureTask 的 run 方法。
            else if (!periodic)
                ScheduledFutureTask.super.run();
                // 調用 runAndReset 方法執行任務。
            else if (ScheduledFutureTask.super.runAndReset()) {
                // 設置下次執行時間
                setNextRunTime();
                // 將任務重新放回工作任務隊列。
                reExecutePeriodic(outerTask);
            }
        }

 

工作任務隊列中的任務對象一旦被工作線程獲取成功后，就會被從隊列中移出。而其他之前阻塞在隊列上，此時競爭到鎖的工作者線程將會嘗試獲取任務隊列中的下一個任務。

調用成功獲取到的 ScheduledFutureTask 的 run 方法，執行業務邏輯以后 將重新計算對象的 delay 時間，再通過 runAndReset 方法將重新計算的后的對象重置回工作任務阻塞隊列中。由於默認實現的 compareTo 方法，
這樣，就實現了線程周期性的執行任務的功能。

總結

ScheduledThreadPoolExecutor 的實現並不復雜，主要是理解有序隊列的操作，以及對 FutureTask 的靈活運用，明白這些后，再看 ScheduledThreadPoolExecutor 就不是難事了。

另外，我們常用的 quartz 就是借用了 ScheduledThreadPoolExecutor 來實現定時任務的執行與調度，只不過提供了一種更友好的方式去表達定時任務的配置方式，為 ScheduledThreadPoolExecutor 需要的數據做了封裝。真正的功能還是圍繞在 ScheduledThreadPoolExecutor上。

 

作者：給你添麻煩了
鏈接：https://www.jianshu.com/p/a39a89d28375
來源：簡書
著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。