1.建立執行緒池相關引數

執行緒池的建立要用ThreadPoolExecutor類的構造方法自定義建立，禁止用Executors的靜態方法建立執行緒池，防止記憶體溢位和建立過多執行緒消耗資源。

corePoolSize: 執行緒池核心執行緒數量，不會自動銷燬，除非設定了引數allowCoreThreadTimeOut=true,那麼即使當執行緒數量小於corePoolSize的時候，當執行緒
空閒時間大於keepAliveTime，也會被回收
maximumPoolSize: 執行緒池能容納的最大執行緒數量 
keepAliveTime: 一般情況下核心執行緒不可回收，非核心執行緒空閒時間大於此時間會被回收
unit: keepAliveTime單位
workQueue: 工作佇列，當執行緒池數量等於corePoolSize的時候，此時任務會先進入到佇列，其他執行緒執行完任務後會從該佇列獲取任務繼續執行
threadFactory: 執行緒工廠，用來建立執行緒池的執行緒同時也可以指定執行緒的名字以及其他屬性
handler: 當執行緒池執行緒已滿執行拒絕策略

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.acc = System.getSecurityManager() == null ?
        null :
    AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}
 

2.核心方法原始碼分析

2.1 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.
     */
    int c = ctl.get();
    // 當執行緒池執行緒數量小於corePoolSize時，直接嘗試新增一個執行緒並把command當做這個執行緒的第一個任務
    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);
        // 到這表明執行緒池此時狀態時Running,然後判斷執行緒池執行緒數量，使其至少有一個執行緒能夠執行任務 
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 佇列已滿新增失敗，嘗試新增執行緒執行任務若新增失敗（執行緒池執行緒數量已達到maximumPoolSize或者執行緒池狀態shutdown等）則執行拒絕策略
    else if (!addWorker(command, false))
        reject(command);
}
 

2.2 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.
        // 第二個與條件分解 rs >= SHUTDOWN && (rs > SHUTDOWN  || firstTask != null || workQueue.isEmpty())
        // 有三種情況執行緒池不增加執行緒
        // 1.執行緒池狀態STOP / TIDYING / TERMINATED
        // 2.執行緒池狀態SHUTDOWN但此時已不接受新提交到執行緒池的任務
        // 3.執行緒池狀態SHUTDOWN此時工作佇列無任務 （由此可見在SHUTDOWN狀態下只要佇列中還存在任務那麼執行緒池還會增加執行緒處理）
        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;
            // 執行緒數量+ 1 跳出迴圈
            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 {
        // 建立Worker工作者
        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());

                // 執行緒池狀態為RUNNING或者 執行緒池狀態為SHUTDOWN但沒有新提交任務（此時增加執行緒是為了處理阻塞佇列中的任務）
                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) {
                // 啟動執行緒開始執行任務 Worker$run()方法
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // 啟動失敗 workers移除執行緒數量-1
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}
 

2.3 runworker詳解

// 上一個方法t.start（）執行後就會執行這個方法
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    // state = 0, 使得執行緒可中斷
    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
            // 若執行緒池狀態為STOP時，確保執行緒有設定中斷狀態，若執行緒池狀態為RUNING和SHUTDOWN，則會清除執行緒的中斷狀態
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                // 前置鉤子方法可重寫
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    // 執行任務 
                    // 由上所知執行緒池為STOP會給執行緒設定中斷狀態，若任務程式碼邏輯有對中斷的相關處理可能會直接丟擲中斷異常
                    // shutdownNow()方法會讓執行緒池放棄佇列中以及正在執行中的任務（若任務中沒有對中斷進行處理則會繼續執行）
                    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 {
                // 清空任務 worker完成任務數+1
                task = null;
                w.completedTasks++;
                // 解鎖
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

2.4 getTask詳解

/**
 * Performs blocking or timed wait for a task, depending on
 * current configuration settings, or returns null if this worker
 * must exit because of any of:
 * 1. There are more than maximumPoolSize workers (due to
 *    a call to setMaximumPoolSize).
 * 2. The pool is stopped.
 * 3. The pool is shutdown and the queue is empty.
 * 4. This worker timed out waiting for a task, and timed-out
 *    workers are subject to termination (that is,
 *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
 *    both before and after the timed wait, and if the queue is
 *    non-empty, this worker is not the last thread in the pool.
 *
 * @return task, or null if the worker must exit, in which case
 *         workerCount is decremented
 */
private Runnable getTask() {
    // 是否超時 keepAliveTime
    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.
        // 如果執行緒池狀態為STOP或者為SHUTDOWN並且任務佇列為空則不在執行任務，直接回收執行緒
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            // 迴圈cas直到成功
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        // 執行緒是否空閒回收
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        // 執行緒池執行緒數超過最大限制或者當前執行緒空閒時間已經超過keepAliveTime
        // 並且執行緒池執行緒數大於1或者佇列為空那麼代表此執行緒可以回收
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            // cas設定執行緒數減1 cas失敗表示執行緒池狀態變化或者其他執行緒先一步回收使得執行緒池執行緒已經減1了
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            // 是否空閒回收， poll限制從佇列獲取任務超時返回
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            // 未取到說明執行緒已空閒keepAliveTime時間 超時可回收
            timedOut = true;
        } catch (InterruptedException retry) {
            // 若執行緒使用take()或poll()且佇列中沒有任務，則當呼叫shutdown()/shutdownNow()時會給執行緒設定中斷狀態 
            // 此時會丟擲中斷異常、並且執行緒池狀態可能已經發生了變化此時開始下一輪迴圈
            timedOut = false;
        }
    }
}

2.5 processWorkerExit詳解

/**
 * Performs cleanup and bookkeeping for a dying worker. Called
 * only from worker threads. Unless completedAbruptly is set,
 * assumes that workerCount has already been adjusted to account
 * for exit.  This method removes thread from worker set, and
 * possibly terminates the pool or replaces the worker if either
 * it exited due to user task exception or if fewer than
 * corePoolSize workers are running or queue is non-empty but
 * there are no workers.
 *
 * @param w the worker
 * @param completedAbruptly if the worker died due to user exception
 */
private void processWorkerExit(Worker w, boolean completedAbruptly) {
    // run方法異常 執行緒數量直接-1
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }

    //嘗試終止執行緒池，設定最終狀態
    tryTerminate();

    int c = ctl.get();
    // 執行緒池狀態為RUNNING或SHUTDONW
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            // min為執行緒池最小不可回收執行緒數
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        // 增加一個執行緒
        addWorker(null, false);
    }
}

2.6 tryTerminate詳解

/**
 * Transitions to TERMINATED state if either (SHUTDOWN and pool
 * and queue empty) or (STOP and pool empty).  If otherwise
 * eligible to terminate but workerCount is nonzero, interrupts an
 * idle worker to ensure that shutdown signals propagate. This
 * method must be called following any action that might make
 * termination possible -- reducing worker count or removing tasks
 * from the queue during shutdown. The method is non-private to
 * allow access from ScheduledThreadPoolExecutor.
 */
final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        // 三種情況什麼都不做直接返回
        // 1.執行緒池狀態還為RUNNING
        // 2.執行緒池狀態為TIDYING或TERMINATED已終止
        // 3.執行緒池狀態為SHUTDOWN但阻塞佇列中還有任務
        // 當狀態為STOP或者為SHUTDOW且佇列為空才會往下執行
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        // 執行緒池是否還存線上程 嘗試中斷一個執行緒，進行傳播
        // 如何理解？ 當一個執行緒從等待中中斷後，getTask（）返回null，後續會執行processWorkerExit，
        // 而processWorkerExit裡的tryTerminate會再次嘗試終止執行緒池或再中斷一個執行緒以達到傳播的目的，妙哉
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        // 到這裡說明執行緒池裡的執行緒都已經回收了，可以嘗試終止執行緒池了
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 執行緒池設定終止狀態了
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

2.7 interruptIdleWorkers詳解

/**
 * Interrupts threads that might be waiting for tasks (as
 * indicated by not being locked) so they can check for
 * termination or configuration changes. Ignores
 * SecurityExceptions (in which case some threads may remain
 * uninterrupted).
 *
 * @param onlyOne If true, interrupt at most one worker. This is
 * called only from tryTerminate when termination is otherwise
 * enabled but there are still other workers.  In this case, at
 * most one waiting worker is interrupted to propagate shutdown
 * signals in case all threads are currently waiting.（向其他正在等待的執行緒傳遞SHUTDOWN關閉訊號）
 * Interrupting any arbitrary thread ensures that newly arriving
 * workers since shutdown began will also eventually exit.
 * To guarantee eventual termination, it suffices to always
 * interrupt only one idle worker, but shutdown() interrupts all
 * idle workers so that redundant workers exit promptly, not
 * waiting for a straggler task to finish.
 */
private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            // 執行緒非中斷並且嘗試獲取到了worker的鎖則給執行緒設定中斷狀態
            // w.tryLock() = true,說明執行緒t沒有執行任務空閒，可能在getTask()中阻塞等待任務
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

2.8 shutdown詳解

/**
 * Initiates an orderly shutdown in which previously submitted
 * tasks are executed, but no new tasks will be accepted.
 * Invocation has no additional effect if already shut down.
 *
 * <p>This method does not wait for previously submitted tasks to
 * complete execution.  Use {@link #awaitTermination awaitTermination}
 * to do that.
 *
 * @throws SecurityException {@inheritDoc} 
 */
// 嘗試停止執行緒池
public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        // 執行緒池狀態設定為SHUTDOWN
        advanceRunState(SHUTDOWN);
        // 中斷所有空閒的執行緒
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

2.9 shutdownNow詳解

/**
 * Attempts to stop all actively executing tasks, halts the
 * processing of waiting tasks, and returns a list of the tasks
 * that were awaiting execution. These tasks are drained (removed)
 * from the task queue upon return from this method.
 *
 * <p>This method does not wait for actively executing tasks to
 * terminate.  Use {@link #awaitTermination awaitTermination} to
 * do that.
 *
 * <p>There are no guarantees beyond best-effort attempts to stop
 * processing actively executing tasks.  This implementation
 * cancels tasks via {@link Thread#interrupt}, so any task that
 * fails to respond to interrupts may never terminate.
 *
 * @throws SecurityException {@inheritDoc}
 */
// 取消執行佇列中的和正在執行中的任務
public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        // 執行緒池狀態設定為STOP
        advanceRunState(STOP);
        // 中斷執行緒池中的所有執行緒，即使是在執行任務中的執行緒
        interruptWorkers();
        // 獲取佇列中的任務並且清空佇列
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    // 嘗試終止執行緒池，若執行緒池為空則終止完成,否則依賴傳播來終止執行緒池
    tryTerminate();
    return tasks;
}

3.問題

3.1 執行緒池被 shutdown 後，還能產生新的執行緒？

執行緒池被shutdown後在阻塞佇列還有任務的情況下會產生新的執行緒，見addWorker方法詳解

3.2 執行緒把任務丟給執行緒池後肯定就馬上返回了?

執行緒把任務丟給執行緒池後不是立即返回，當執行緒池執行緒數量小於核心執行緒數的時候會接著建立一個執行緒並直接執行這個提交到執行緒池的任務，當執行緒池執行緒數量等於核心執行緒數會將任務新增到阻塞佇列（若新增失敗則使用拒絕策略，新增成功會判斷當前執行緒池是否有執行緒存在，不存在則建立一個執行緒），若任務佇列已滿則嘗試建立一個非核心執行緒執行此任務。參考execute方法

3.3 執行緒池裡的執行緒異常後會再次新增執行緒嗎，如何捕獲這些執行緒丟擲的異常?

執行緒池裡的執行緒執行異常會丟擲異常且會將該worker移除，若執行緒池狀態不為STOP則會再次新增執行緒。

1.在run方法中捕獲

2.通過submit提交任務獲取future，之後呼叫get()方法若執行緒有丟擲異常則會捕獲到

3.重寫afterExecute()方法，此方法會以runnable和throwable為入參

4.通過Thread的uncaughtExceptionHander處理

3.4 執行緒池的大小如何設定，如何動態設定執行緒池的引數

setMaximumPoolSize 若當前執行緒池執行緒數量大於設定的maximumPoolSize 會嘗試中斷執行緒，達到回收多餘執行緒的目的

setCorePoolSize 若當前執行緒池執行緒數量大於設定的corePoolSize會嘗試中斷執行緒，若設定值大於原先的corePoolSize則會根據佇列中的任務建立合適數量的執行緒來執行任務

3.5 阿里 Java 程式碼規範為什麼不允許使用 Executors 快速建立執行緒池？

newFixedThreadPool和newSingleThreadExecutor建立workQueue LinkedBlockingQueue未宣告大小相當於建立無界佇列,若任務數量過多新增到佇列中可能會導致OOM

newCachedThreadPool和newScheduledThreadPool最大執行緒數設定為Integer.MAX_VALUE也可能導致OOM

3.6 如何優雅關閉執行緒池

呼叫shutdown()或shutdownNow()方法關閉執行緒池後使用awaitTermination方法等待執行緒池執行緒和佇列任務清空變為TERMINATED狀態

3.7 使用執行緒池應該避免哪些問題，能否簡單說下執行緒池的最佳實踐？

1.執行緒池執行的任務應該是相互獨立的，如果都在一個執行緒池裡執行可能會導致死鎖

2.核心任務與非核心任務最好能用多個執行緒池隔離開來，非核心任務可能過多導致核心任務堆積在佇列中無法及時執行，影響業務

3.執行緒池各個引數很難一次性確定，可以新增告警，比如三分鐘內佇列任務數都是滿的情況下觸發告警，支援動態調整修改執行緒池的核心執行緒數和最大執行緒數。

4. 結語   

本篇文章簡要分析了執行緒池ThreadPoolExecutor中比較重要的幾個方法以及對幾個常見問題的理解，如果理解的有問題歡迎讀者大佬指出討論，謝謝~