【死磕Java併發】—–J.U.C之執行緒池：ThreadPoolExecutor

ThreadPoolExecutor提供實現：

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

執行流程如下：

1. 如果執行緒池當前執行緒數小於corePoolSize，則呼叫addWorker建立新執行緒執行任務，成功傳回true，失敗執行步驟2。

2. 如果執行緒池處於RUNNING狀態，則嘗試加入阻塞佇列，如果加入阻塞佇列成功，則嘗試進行Double Check，如果加入失敗，則執行步驟3。

3. 如果執行緒池不是RUNNING狀態或者加入阻塞佇列失敗，則嘗試建立新執行緒直到maxPoolSize，如果失敗，則呼叫reject()方法執行相應的拒絕策略。

在步驟2中如果加入阻塞佇列成功了，則會進行一個Double Check的過程。Double Check過程的主要目的是判斷加入到阻塞隊裡中的執行緒是否可以被執行。如果執行緒池不是RUNNING狀態，則呼叫remove()方法從阻塞佇列中刪除該任務，然後呼叫reject()方法處理任務。否則需要確保還有執行緒執行。

addWorker當執行緒中的當前執行緒數小於corePoolSize，則呼叫addWorker()建立新執行緒執行任務，當前執行緒數則是根據ctl變數來獲取的，呼叫workerCountOf(ctl)獲取低29位即可：

    private static int workerCountOf(int c)  { return c & CAPACITY; }

addWorker(Runnable firstTask, boolean core)方法用於建立執行緒執行任務，原始碼如下：

    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            // 獲取當前執行緒狀態
            int rs = runStateOf(c);
            if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                            firstTask == null &&
                            ! workQueue.isEmpty()))
                return false;
            // 內層迴圈，worker + 1
            for (;;) {
                // 執行緒數量
                int wc = workerCountOf(c);
                // 如果當前執行緒數大於執行緒最大上限CAPACITY  return false
                // 若core == true，則與corePoolSize 比較，否則與maximumPoolSize ，大於 return false
                if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                // worker + 1,成功跳出retry迴圈
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                // CAS add worker 失敗，再次讀取ctl
                c = ctl.get();
                // 如果狀態不等於之前獲取的state，跳出內層迴圈，繼續去外層迴圈判斷
                if (runStateOf(c) != rs)
                    continue retry;
            }
        }
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            // 新建執行緒：Worker
            w = new Worker(firstTask);
            // 當前執行緒
            final Thread t = w.thread;
            if (t != null) {
                // 獲取主鎖：mainLock
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // 執行緒狀態
                    int rs = runStateOf(ctl.get());
                    // rs < SHUTDOWN ==> 執行緒處於RUNNING狀態
                    // 或者執行緒處於SHUTDOWN狀態，且firstTask == null（可能是workQueue中仍有未執行完成的任務，建立沒有初始任務的worker執行緒執行）
                    if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                        // 當前執行緒已經啟動，丟擲異常
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        // workers是一個HashSet
                        workers.add(w);
                        // 設定最大的池大小largestPoolSize，workerAdded設定為true
                        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;
    }

1. 判斷當前執行緒是否可以新增任務，如果可以則進行下一步，否則return false；

2. rs >= SHUTDOWN ，表示當前執行緒處於SHUTDOWN ，STOP、TIDYING、TERMINATED狀態

3. rs == SHUTDOWN , firstTask != null時不允許新增執行緒，因為執行緒處於SHUTDOWN 狀態，不允許新增任務

4. rs == SHUTDOWN , firstTask == null，但workQueue.isEmpty() == true，不允許新增執行緒，因為firstTask == null是為了新增一個沒有任務的執行緒然後再從workQueue中獲取任務的，如果workQueue == null，則說明新增的任務沒有任何意義。

5. 內嵌迴圈，透過CAS worker + 1

6. 獲取主鎖mailLock，如果執行緒池處於RUNNING狀態獲取處於SHUTDOWN狀態且 firstTask == null，則將任務新增到workers Queue中，然後釋放主鎖mainLock，然後啟動執行緒，然後return true，如果中途失敗導致workerStarted= false，則呼叫addWorkerFailed()方法進行處理。

在這裡需要好好理論addWorker中的引數，在execute()方法中，有三處呼叫了該方法：
第一次：workerCountOf(c) < corePoolSize ==> addWorker(command, true)，這個很好理解，當然執行緒池的執行緒數量小於 corePoolSize ，則新建執行緒執行任務即可，在執行過程core == true，內部與corePoolSize比較即可。
第二次：加入阻塞佇列進行Double Check時，else if (workerCountOf(recheck) == 0) ==>addWorker(null, false)。如果執行緒池中的執行緒==0，按照道理應該該任務應該新建執行緒執行任務，但是由於已經該任務已經新增到了阻塞佇列，那麼就在執行緒池中新建一個空執行緒，然後從阻塞佇列中取執行緒即可。
第三次：執行緒池不是RUNNING狀態或者加入阻塞佇列失敗：else if (!addWorker(command, false))，這裡core == fase，則意味著是與maximumPoolSize比較。

在新建執行緒執行任務時，將講Runnable包裝成一個Worker，Woker為ThreadPoolExecutor的內部類

Woker內部類

Woker的原始碼如下：

    private final class Worker extends AbstractQueuedSynchronizer
            implements Runnable {
        private static final long serialVersionUID = 6138294804551838833L;
        // task 的thread
        final Thread thread;
        // 執行的任務task
        Runnable firstTask;
        volatile long completedTasks;
        Worker(Runnable firstTask) {
            //設定AQS的同步狀態private volatile int state，是一個計數器，大於0代表鎖已經被獲取
            setState(-1);
            this.firstTask = firstTask;
            // 利用ThreadFactory和 Worker這個Runnable建立的執行緒物件
            this.thread = getThreadFactory().newThread(this);
        }
        // 任務執行
        public void run() {
            runWorker(this);
        }
    }

從Worker的原始碼中我們可以看到Woker繼承AQS，實現Runnable介面，所以可以認為Worker既是一個可以執行的任務，也可以達到獲取鎖釋放鎖的效果。這裡繼承AQS主要是為了方便執行緒的中斷處理。這裡註意兩個地方：建構式、run()。建構式主要是做三件事：1.設定同步狀態state為-1，同步狀態大於0表示就已經獲取了鎖，2.設定將當前任務task設定為firstTask，3.利用Worker本身物件this和ThreadFactory建立執行緒物件。

當執行緒thread啟動（呼叫start()方法）時，其實就是執行Worker的run()方法，內部呼叫runWorker()。

runWorker

    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) {
                // worker 獲取鎖
                w.lock();
                // 確保只有當執行緒是stoping時，才會被設定為中斷，否則清楚中斷標示
                // 如果執行緒池狀態 >= STOP ,且當前執行緒沒有設定中斷狀態，則wt.interrupt()
                // 如果執行緒池狀態 < STOP，但是執行緒已經中斷了，再次判斷執行緒池是否 >= STOP，如果是 wt.interrupt()
                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;
                    // 完成任務數 + 1
                    w.completedTasks++;
                    // 釋放鎖
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

執行流程

1. 根據worker獲取要執行的任務task，然後呼叫unlock()方法釋放鎖，這裡釋放鎖的主要目的在於中斷，因為在new Worker時，設定的state為-1，呼叫unlock()方法可以將state設定為0，這裡主要原因就在於interruptWorkers()方法只有在state >= 0時才會執行；

2. 透過getTask()獲取執行的任務，呼叫task.run()執行，當然在執行之前會呼叫worker.lock()上鎖，執行之後呼叫worker.unlock()放鎖；

3. 在任務執行前後，可以根據業務場景自定義beforeExecute() 和 afterExecute()方法，則兩個方法在ThreadPoolExecutor中是空實現；

4. 如果執行緒執行完成，則會呼叫getTask()方法從阻塞佇列中獲取新任務，如果阻塞佇列為空，則根據是否超時來判斷是否需要阻塞；

5. task == null或者丟擲異常（beforeExecute()、task.run()、afterExecute()均有可能）導致worker執行緒終止，則呼叫processWorkerExit()方法處理worker退出流程。

getTask()

    private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?
        for (;;) {
            // 執行緒池狀態
            int c = ctl.get();
            int rs = runStateOf(c);
            // 執行緒池中狀態 >= STOP 或者 執行緒池狀態 == SHUTDOWN且阻塞佇列為空，則worker - 1，return null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
            int wc = workerCountOf(c);
            // 判斷是否需要超時控制
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
            try {
                // 從阻塞佇列中獲取task
                // 如果需要超時控制，則呼叫poll()，否則呼叫take()
                Runnable r = timed ?
                        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                        workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

timed == true，呼叫poll()方法，如果在keepAliveTime時間內還沒有獲取task的話，則傳回null，繼續迴圈。timed == false，則呼叫take()方法，該方法為一個阻塞方法，沒有任務時會一直阻塞掛起，直到有任務加入時對該執行緒喚醒，傳回任務。

在runWorker()方法中，無論最終結果如何，都會執行processWorkerExit()方法對worker進行退出處理。

processWorkerExit()

    private void processWorkerExit(Worker w, boolean completedAbruptly) {
        // true：使用者執行緒執行異常,需要扣減
        // false：getTask方法中扣減執行緒數量
        if (completedAbruptly)
            decrementWorkerCount();
        // 獲取主鎖
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks;
            // 從HashSet中移出worker
            workers.remove(w);
        } finally {
            mainLock.unlock();
        }
        // 有worker執行緒移除，可能是最後一個執行緒退出需要嘗試終止執行緒池
        tryTerminate();
        int c = ctl.get();
        // 如果執行緒為running或shutdown狀態，即tryTerminate()沒有成功終止執行緒池，則判斷是否有必要一個worker
        if (runStateLessThan(c, STOP)) {
            // 正常退出，計算min：需要維護的最小執行緒數量
            if (!completedAbruptly) {
                // allowCoreThreadTimeOut 預設false：是否需要維持核心執行緒的數量
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                // 如果min ==0 或者workerQueue為空，min = 1
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;
                // 如果執行緒數量大於最少數量min，直接傳回，不需要新增執行緒
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            // 新增一個沒有firstTask的worker
            addWorker(null, false);
        }
    }

首先completedAbruptly的值來判斷是否需要對執行緒數-1處理，如果completedAbruptly == true，說明在任務執行過程中出現了異常，那麼需要進行減1處理，否則不需要，因為減1處理在getTask()方法中處理了。然後從HashSet中移出該worker，過程需要獲取mainlock。然後呼叫tryTerminate()方法處理，該方法是對最後一個執行緒退出做終止執行緒池動作。如果執行緒池沒有終止，那麼執行緒池需要保持一定數量的執行緒，則透過addWorker(null,false)新增一個空的執行緒。

addWorkerFailed()

在addWorker()方法中，如果執行緒t==null，或者在add過程出現異常，會導致workerStarted == false，那麼在最後會呼叫addWorkerFailed()方法：

    private void addWorkerFailed(Worker w) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 從HashSet中移除該worker
            if (w != null)
                workers.remove(w);
            // 執行緒數 - 1
            decrementWorkerCount();
            // 嘗試終止執行緒
            tryTerminate();
        } finally {
            mainLock.unlock();
        }
    }

整個邏輯顯得比較簡單。

tryTerminate()

當執行緒池涉及到要移除worker時候都會呼叫tryTerminate()，該方法主要用於判斷執行緒池中的執行緒是否已經全部移除了，如果是的話則關閉執行緒池。

    final void tryTerminate() {
        for (;;) {
            int c = ctl.get();
            // 執行緒池處於Running狀態
            // 執行緒池已經終止了
            // 執行緒池處於ShutDown狀態，但是阻塞佇列不為空
            if (isRunning(c) ||
                    runStateAtLeast(c, TIDYING) ||
                    (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;
            // 執行到這裡，就意味著執行緒池要麼處於STOP狀態，要麼處於SHUTDOWN且阻塞佇列為空
            // 這時如果執行緒池中還存在執行緒，則會嘗試中斷執行緒
            if (workerCountOf(c) != 0) {
                // /執行緒池還有執行緒，但是佇列沒有任務了，需要中斷喚醒等待任務的執行緒
                // （runwoker的時候首先就透過w.unlock設定執行緒可中斷，getTask最後面的catch處理中斷）
                interruptIdleWorkers(ONLY_ONE);
                return;
            }
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 嘗試終止執行緒池
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                        terminated();
                    } finally {
                        // 執行緒池狀態轉為TERMINATED
                        ctl.set(ctlOf(TERMINATED, 0));
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
        }
    }

在關閉執行緒池的過程中，如果執行緒池處於STOP狀態或者處於SHUDOWN狀態且阻塞佇列為null，則執行緒池會呼叫interruptIdleWorkers()方法中斷所有執行緒，註意ONLY_ONE== true，表示僅中斷一個執行緒。

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();
        }
    }

onlyOne==true僅終止一個執行緒，否則終止所有執行緒。

執行緒終止

執行緒池ThreadPoolExecutor提供了shutdown()和shutDownNow()用於關閉執行緒池。

shutdown()：按過去執行已提交任務的順序發起一個有序的關閉，但是不接受新任務。

shutdownNow() :嘗試停止所有的活動執行任務、暫停等待任務的處理，並傳回等待執行的任務串列。

shutdown

    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            // 推進執行緒狀態
            advanceRunState(SHUTDOWN);
            // 中斷空閑的執行緒
            interruptIdleWorkers();
            // 交給子類實現
            onShutdown();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

shutdownNow

    public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(STOP);
            // 中斷所有執行緒
            interruptWorkers();
            // 傳回等待執行的任務串列
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
    }

與shutdown不同，shutdownNow會呼叫interruptWorkers()方法中斷所有執行緒。

    private void interruptWorkers() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers)
                w.interruptIfStarted();
        } finally {
            mainLock.unlock();
        }
    }

同時會呼叫drainQueue()方法傳回等待執行到任務串列。

    private List<Runnable> drainQueue() {
        BlockingQueue<Runnable> q = workQueue;
        ArrayList<Runnable> taskList = new ArrayList<Runnable>();
        q.drainTo(taskList);
        if (!q.isEmpty()) {
            for (Runnable r : q.toArray(new Runnable[0])) {
                if (q.remove(r))
                    taskList.add(r);
            }
        }
        return taskList;
    }

END

chenssy

贊賞

長按二維碼向我轉賬

受蘋果公司新規定影響，微信 iOS 版的贊賞功能被關閉，可透過二維碼轉賬支援公眾號。

閱讀原文


微信掃一掃
使用小程式
即將開啟""小程式
取消
開啟