ThreadPoolExecutor線程池源碼分析

時間 2019-12-05

標籤 threadpoolexecutor 線程源碼分析简体版

原文原文鏈接

瞭解ThreadPoolExecutor

先看一下線程池類的類圖關係：java

Executor接口

Executor做者描述的是Executor提供了一種解耦方式將任務的提交和任務以何種技術執行分離；
Executor接口只有一個方法：安全

void execute(Runnable command);

execute方法接收一個Runnable對象，方法的描述是在將來的某個時間執行command。不論是在一個新的線程中執行，仍是在線程池中執行，甚至在調用者線程中當即執行。異步

ExecutorService接口

ExecutorService繼承了Executor接口，ExecutorService能夠被關閉，關閉之後再也不接收新的任務。ExecutorService提供了兩個不一樣的方法關閉ExecutorService。shutdown方法會等待以前還未執行的任務執行完畢再關閉，而shutdownNow則不會再啓動新的任務，還會中斷正在執行的任務。一旦關閉後，ExecutorService就不會有正在執行的任務，也不會有等待被執行的任務，更不會有新的任務被提交。ExecutorService關閉後應該處理好一些資源的回收。ide

ThreadPoolExecutor

線程池技術旨在解決兩個不一樣的問題：函數

在處理大量異步任務時能夠提升性能，由於減小了線程的銷燬，新建，切換等消耗性能的操做。
線程池還有能力統一管理，調度，監控，調優線程等，還提供了一下基本的統計，好比已完成的任務數量。

重要的狀態和狀態判斷的方法

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//高3位和低29位分別表示狀態和線程數
private static final int COUNT_BITS = Integer.SIZE - 3;
//1左移29位減一獲得低29位都是1，即線程的最大數量，大概5億多
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;//111
private static final int SHUTDOWN   =  0 << COUNT_BITS;//000
private static final int STOP       =  1 << COUNT_BITS;//001
private static final int TIDYING    =  2 << COUNT_BITS;//010
private static final int TERMINATED =  3 << COUNT_BITS;//011

// Packing and unpacking ctl
//得到狀態
private static int runStateOf(int c)     { return c & ~CAPACITY; }
//得到線程數量
private static int workerCountOf(int c)  { return c & CAPACITY; }
//經過狀態和線程數量組裝ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }

/*
 * Bit field accessors that don't require unpacking ctl.
 * These depend on the bit layout and on workerCount being never negative.
 */
//c狀態是否小於s狀態
private static boolean runStateLessThan(int c, int s) {
    return c < s;
}
//c狀態是否大於等於s狀態
private static boolean runStateAtLeast(int c, int s) {
    return c >= s;
}
//線程池是不是運行狀態
private static boolean isRunning(int c) {
    return c < SHUTDOWN;
}

整個類最重要的一個狀態標誌ctl是一個AtomicInteger，它包含了兩個字段的含義。workerCount線程數量，runState線程池的狀態。
這一個字段是如何包含兩個字段的含義的呢，Doug Lea大牛使用了一個int的32位bits的高三位保存了狀態值，低29位保存了線程數量。oop

其中五個狀態：
RUNNING：接收新的任務，處理隊列中的任務；
SHUTDOWN：不接收新的任務，但處理隊列中的任務；
STOP：不接收新的任務，不處理隊列中的任務，中斷正在執行的任務；
TIDYING：全部任務都終止，線程數爲0，線程過分到TIDYING時會調用terminated鉤子方法；
TERMINATED：terminated執行完畢；性能

狀態之間的轉換：
RUNNING -> SHUTDOWN：調用shutdown方法；
(RUNNING or SHUTDOWN) -> STOP：調用shutdownNow方法；
SHUTDOWN -> TIDYING：當線程池和任務隊列都爲空；
STOP -> TIDYING：當線程池爲空；
TIDYING -> TERMINATED：當terminated方法執行完畢；ui

Worker介紹

Worker類主要包含了線程運行任務時的終端控制狀態，同時還有一些少許的信息記錄。Worker適時的繼承了AQS，讓線程在任務執行之間獲取鎖和釋放鎖變得簡單。這確保了中斷是喚醒一個等待任務的線程，而不是中斷一個正在運行的任務線程。this

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;

    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;

    /**
     * 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);
    }

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

    // Lock methods
    //
    // The value 0 represents the unlocked state.
    // The value 1 represents the locked state.

    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }

    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

Worker繼承了AQS，實現了Runnable接口；在構造函數中，初始化了它的第一次仍無，使用threadFactory建立一個新的線程；
Worker繼承AQS，目的是想使用獨佔鎖來表示線程是否正在執行任務，Worker的線程獲取了獨佔鎖就說明它在執行任務，不能被中斷。從tryAcquire方法能夠看出，它實現的是不可重入鎖，由於是否得到鎖在這裏表示一個狀態，若是能夠重入的話，獨佔鎖就失去了只表示一個狀態的含義。在構造函數初始化時，Worker將state設置爲-1，由於在tryAcquire中CAS操做compareAndSetState(0, 1)，表示state在-1時不能被中斷。在runWorker中將state設置爲0.atom

ThreadPooleExecutor構造方法

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

說明一下各參數的含義：
corePoolSize：核心線程數量，即便線程是空閒的也保持在線程池中，除非allowCoreThreadTimeOut參數被設置；

maximumPoolSize：最大線程數量；

keepAliveTime：當線程數量超過核心線程數量時，超出的空閒線程等待新任務的最大時長；

unit：時間單位；

workQueue：存放將要被執行的任務的隊列；

threadFactory：建立線程的線程工廠；

handler：當任務隊列滿且沒有空閒的線程時處理任務的handler，線程池提供了四種策略：

AbortPolicy：直接拋出異常，默認；
CallerRunsPolicy：使用調用者的線程執行；
DiscardOldestPolicy：拋棄隊列最前的任務，執行當前任務；
DiscardPolicy：直接丟棄任務；

這些參數對整個線程池運行很是重要；

execute方法

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
    //獲取ctl
    int c = ctl.get();
    //若是線程數小於核心線程數
    if (workerCountOf(c) < corePoolSize) {
        //添加線程並執行任務
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //線程數大於核心線程數
    //若是線程池running狀態且添加任務到隊列成功
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        //若是線程池不是運行狀態，隊列移除任務，使用拒絕策略處理任務
        if (! isRunning(recheck) && remove(command))
            reject(command);
        //若是這時線程數爲0，添加任務
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    //隊列滿，添加線程失敗，使用拒絕策略處理任務
    else if (!addWorker(command, false))
        reject(command);
}

在線程池添

數量若是小於核心線程數，則添加新的線程並執行當前任務，不然判斷若是隊列是否未滿，則添加當前任務到隊列，不然判斷線程數量若是小於最大線程數，則添加新的線程並執行，不然使用拒絕策略處理當前任務。

addWorker方法

addWorker方法主要是添加線程並執行任務：

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        //獲取線程池運行狀態
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        //若是運行狀態大於等於SHUTDOWN，再也不接受新的任務，返回false
        //若是運行狀態等於SHUTDOWN且firstTask不爲空，繼續執行下去，若是firstTask爲空，queue爲空，返回false，不然繼續執行；只要SHUTDOWN狀態下還有任務在，就須要往下執行，可能須要新建worker執行
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            //得到線程數量
            int wc = workerCountOf(c);
            //若是線程數量大於容量或者當core爲true時wc大於等於核心線程數，當core爲falsewc大於等於最大線程數量時，返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            //CAS線程數加一，成功則中斷循環
            if (compareAndIncrementWorkerCount(c))
                break retry;
            //若是CAS失敗，從新獲取ctl，線程池運行狀態沒變的話繼續loop
            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 {
        //新建一個worker
        w = new Worker(firstTask);
        //能獲得worker的thread
        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());

                //若是rs是RUNNING或者SHUTDOWN且firstTask爲null
                //由於SHUTDOWN時還須要執行queue中的任務
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    //往線程池中添加worker
                    workers.add(w);
                    int s = workers.size();
                    //記錄線程池出現的最大線程數量
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                //啓動worker
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

worker的run方法調用的是runWorker；

runWorker方法

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    //保存worker的第一個任務
    Runnable task = w.firstTask;
    //清空worker的第一個任務
    w.firstTask = null;
    //這裏將worker的state設置爲0，容許中斷
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        //若是task爲空，則從隊列中獲取任務
        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
            //若是當前狀態大於等於STOP要保持當前線程中斷
            //若是當前線程小於STOP即RUNNING或者SHUTDOWN，調用Thread.interrupted()清空中斷標誌，若是這時調用了shutdownNow狀態爲STOP，仍是要保持中斷狀態
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                //執行任務前作的事
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    //執行任務
                    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;
                //worker的完成任務數量加一，此時是線程安全的
                w.completedTasks++;
                //釋放鎖
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        //線程退出
        processWorkerExit(w, completedAbruptly);
    }
}

每一個task在調用runWorker後會一直循環執行任務，直到queue中沒有任務了，循環結束，worker生命週期結束。

getTask

上面runWorker時調用了getTask去獲取隊列中的任務，下面咱們看一下這個方法：

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        //若是rs大於等於SHUTDOWN，當RS大於等於STOP說明線程池已經不處理隊列中的任務了，當rs爲SHUTDOWN時，若是隊列是空的，返回null
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            //線程數減一
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        //是否超時控制，allowCoreThreadTimeOut默認false，表明不容許核心線程超時，對於超出核心線程的線程須要控制超時
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        //當線程數大於最大線程數，或者須要超時控制且上次獲取任務超時
        //且線程數大於1或者隊列爲空，嘗試將線程數減一併返回null
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            //失敗重試
            continue;
        }

        try {
            //當須要超時控制時，在keepAliveTime時間內沒有獲取到任務的話返回null，不然調用take獲取任務，此時線程時阻塞的
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

getTask方法在線程數量大於核心線程數時會判斷在獲取task時進行超時判斷（poll），超時返回null這時getTask返回null，那當前worker的loop結束即run方法結束，線程生命週期結束。而核心線程則會調用take方法，當沒有任務時會阻塞。

processWorkerExit

runTask方法最後會調用processWorkerExit方法進行一些cleanup工做。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    //completedAbruptly爲true時表明發生了異常，線程數減一
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //統計完成任務數
        completedTaskCount += w.completedTasks;
        //線程池移除當前worker
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    // 根據線程池狀態進行判斷是否結束線程池
    tryTerminate();

    int c = ctl.get();
    //當線程池狀態爲RUNNING或者SHUTDOWN時
    //若是發生異常，從新加入一個worker replacement
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            //當allowCoreThreadTimeOut爲true，最少要一個worker
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            //當線程數大於等於最少須要的線程數，則不須要add新的worker
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

tryTerminate方法

上面咱們跳過了tryTerminate方法，該方法判斷是否要結束線程池，這裏看一下

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        //當線程池狀態時RUNNING或者已經TIDYING或者已經TERMINATED或者SHUTDOWN且還有任務沒有被執行，直接返回
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        // 若是線程數不爲0，則中斷一個空閒的工做線程
        if (workerCountOf(c) != 0) { // Eligible to terminate
            //workQueue.take()時若是queue一直爲空的話，線程會一直阻塞
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //若是狀態設置成功爲TIDYING，調用勾子方法terminated，該方法留給了子類實現
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    //設置狀態爲TERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

interruptIdleWorkers

上面說爲了當隊列一直爲空的時候，核心線程會一直阻塞，因此調用了interruptIdleWorkers，咱們看一下執行了什麼：

private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

遍歷線程池中全部的線程，若線程沒有被中斷tryLock成功，就中斷該線程，LockSupport.park()能響應中斷信號，阻塞的線程被中斷喚醒。