摘要：原理剖析第篇工作原理分析一大致介紹相信大家都用過線程池，對該類應該一點都不陌生了我們之所以要用到線程池，線程池主要用來解決線程生命周期開銷問題和資源不足問題我們通過對多個任務重用線程以及控制線程池的數目可以有效防止資源不足的情況本章節就著

-

1、相信大家都用過線程池，對該類ThreadPoolExecutor應該一點都不陌生了；
2、我們之所以要用到線程池，線程池主要用來解決線程生命周期開銷問題和資源不足問題；
3、我們通過對多個任務重用線程以及控制線程池的數目可以有效防止資源不足的情況；
4、本章節就著重和大家分享分析一下JDK8的ThreadPoolExecutor核心類，看看線程池是如何工作的；

1、四個構造器：
    // 構造器一
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
    
    // 構造器二
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }

    // 構造器三
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
    }

    // 構造器四
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue 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;
    }

2、通過仔細查看構造器代碼，發現最終都是調用構造器四，緊接著賦值了一堆的字段，接下來我們先看看這些字段是什么含義；

1、corePoolSize：核心運行的線程池數量大小，當線程數量超過該值時，就需要將超過該數量值的線程放到等待隊列中；

2、maximumPoolSize：線程池最大能容納的線程數(該數量已經包含了corePoolSize數量)，當線程數量超過該值時，則會拒絕執行處理策略；

3、workQueue：等待隊列，當達到corePoolSize的時候，就將新加入的線程追加到workQueue該等待隊列中；
              當然BlockingQueue類也是一個抽象類，也有很多子類來實現不同的隊列等待；
              
              一般來說，阻塞隊列有一下幾種，ArrayBlockingQueue;LinkedBlockingQueue/SynchronousQueue/ArrayBlockingQueue/
              PriorityBlockingQueue使用較少，一般使用LinkedBlockingQueue和Synchronous。

4、keepAliveTime：表示線程沒有任務執行時最多保持多久存活時間，默認情況下當線程數量大于corePoolSize后keepAliveTime才會起作用
                  并生效，一旦線程池的數量小于corePoolSize后keepAliveTime又不起作用了；
                  
                  但是如果調用了allowCoreThreadTimeOut(boolean)方法，在線程池中的線程數不大于corePoolSize時，
                  keepAliveTime參數也會起作用，直到線程池中的線程數為0；

5、threadFactory：新創建線程出生的地方；

6、handler：拒絕執行處理抽象類，就是說當線程池在一些場景中，不能處理新加入的線程任務時，會通過該對象處理拒絕策略；
            該對象RejectedExecutionHandler有四個實現類，即四種策略，讓我們有選擇性的在什么場景下該怎么使用拒絕策略；
            策略一( CallerRunsPolicy )：只要線程池沒關閉，就直接用調用者所在線程來運行任務；
            策略二( AbortPolicy )：默認策略，直接拋出RejectedExecutionException異常；
            策略三( DiscardPolicy )：執行空操作，什么也不干，拒絕任務后也不做任何回應；
            策略四( DiscardOldestPolicy )：將隊列中存活最久的那個未執行的任務拋棄掉，然后將當前新的線程放進去；
   
7、largestPoolSize：變量記錄了線程池在整個生命周期中曾經出現的最大線程個數；

8、allowCoreThreadTimeOut：當為true時，和弦線程也有超時退出的概念一說；

1、AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
   // 原子變量值，是一個復核類型的成員變量，是一個原子整數，借助高低位包裝了兩個概念：
   
2、int COUNT_BITS = Integer.SIZE - 3;
   // 偏移位數，常量值29，之所以偏移29，目的是將32位的原子變量值ctl的高3位設置為線程池的狀態，低29位作為線程池大小數量值；
    
3、int CAPACITY   = (1 << COUNT_BITS) - 1;
   // 線程池的最大容量值；
    
4、線程池的狀態，原子變量值ctl的高三位：
   int RUNNING    = -1 << COUNT_BITS; // 接受新任務，并處理隊列任務
   int SHUTDOWN   =  0 << COUNT_BITS; // 不接受新任務，但會處理隊列任務
   int STOP       =  1 << COUNT_BITS; // 不接受新任務，不會處理隊列任務，而且會中斷正在處理過程中的任務
   int TIDYING    =  2 << COUNT_BITS; // 所有的任務已結束，workerCount為0，線程過渡到TIDYING狀態，將會執行terminated()鉤子方法
   int TERMINATED =  3 << COUNT_BITS; // terminated()方法已經完成
   
5、HashSet workers = new HashSet(); 
   // 存放工作線程的線程池；

1、public void execute(Runnable command)
   // 提交任務，添加Runnable對象到線程池，由線程池調度執行

2、private static int workerCountOf(int c)  { return c & CAPACITY; }
   // c & 高3位為0，低29位為1的CAPACITY，用于獲取低29位的線程數量

3、private boolean addWorker(Runnable firstTask, boolean core)
   // 添加worker工作線程，根據邊界值來決定是否創建新的線程

4、private static boolean isRunning(int c)
   // c通常一般為ctl，ctl值小于0，則處于可以接受新任務狀態

5、final void reject(Runnable command) 
   // 拒絕執行任務方法，當線程池在一些場景中，不能處理新加入的線程時，會通過該對象處理拒絕策略；
   
6、final void runWorker(Worker w)
   // 該方法被Worker工作線程的run方法調用，真正核心處理Runable任務的方法

7、private static int runStateOf(int c)     { return c & ~CAPACITY; }
   // c & 高3位為1，低29位為0的~CAPACITY，用于獲取高3位保存的線程池狀態
   
8、public void shutdown()
   // 不會立即終止線程池，而是要等所有任務緩存隊列中的任務都執行完后才終止，但再也不會接受新的任務

9、public List shutdownNow()
   // 立即終止線程池，并嘗試打斷正在執行的任務，并且清空任務緩存隊列，返回尚未執行的任務
   
10、private void processWorkerExit(Worker w, boolean completedAbruptly)
   // worker線程退出   

1、execute源碼：
   /**
     * Executes the given task sometime in the future.  The task
     * may execute in a new thread or in an existing pooled thread.
     *
     * If the task cannot be submitted for execution, either because this
     * executor has been shutdown or because its capacity has been reached,
     * the task is handled by the current {@code RejectedExecutionHandler}.
     *
     * @param command the task to execute
     * @throws RejectedExecutionException at discretion of
     *         {@code RejectedExecutionHandler}, if the task
     *         cannot be accepted for execution
     * @throws NullPointerException if {@code command} is null
     */
    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(); // 獲取原子計數值最新值
        if (workerCountOf(c) < corePoolSize) { // 判斷當前線程池數量是否小于核心線程數量
            if (addWorker(command, true)) // 嘗試添加command任務到核心線程
                return;
            c = ctl.get(); // 重新獲取當前線程池狀態值，為后面的檢查做準備。
        }
        // 執行到此，說明核心線程任務數量已滿，新添加的線程入等待隊列，這個熟練是大于corePoolSize且小于maximumPoolSize
        if (isRunning(c) && workQueue.offer(command)) { // 如果線程池處于可接受任務狀態，嘗試添加到等待隊列
            int recheck = ctl.get(); // 雙重校驗
            if (! isRunning(recheck) && remove(command)) // 如果線程池突然不可接受任務，則嘗試移除該command任務
                reject(command); // 不可接受任務且成功從等待隊列移除任務，則執行拒絕策略操作，通過策略告訴調用方任務入隊情況
            else if (workerCountOf(recheck) == 0) // 如果此刻線程數量為0的話將沒有Worker執行新的task，所以增加一個Worker
                addWorker(null, false); // 添加一個Worker
        }
        // 執行到此，說明添加任務等待隊列已滿，所以嘗試添加一個Worker
        else if (!addWorker(command, false)) // 如果添加失敗的話，那么拒絕此線程任務添加
            reject(command); // 拒絕此線程任務添加
    }
    
2、小結：
    ? 如果線程池中的線程數量 < corePoolSize，就創建新的線程來執行新添加的任務；
    ? 如果線程池中的線程數量 >= corePoolSize，但隊列workQueue未滿，則將新添加的任務放到workQueue中；
    ? 如果線程池中的線程數量 >= corePoolSize，且隊列workQueue已滿，但線程池中的線程數量 < maximumPoolSize，則會創建新的線程來處理被添加的任務；
    ? 如果線程池中的線程數量 = maximumPoolSize，就用RejectedExecutionHandler來執行拒絕策略；

1、addWorker源碼：
   /**
     * Checks if a new worker can be added with respect to current
     * pool state and the given bound (either core or maximum). If so,
     * the worker count is adjusted accordingly, and, if possible, a
     * new worker is created and started, running firstTask as its
     * first task. This method returns false if the pool is stopped or
     * eligible to shut down. It also returns false if the thread
     * factory fails to create a thread when asked.  If the thread
     * creation fails, either due to the thread factory returning
     * null, or due to an exception (typically OutOfMemoryError in
     * Thread.start()), we roll back cleanly.
     *
     * @param firstTask the task the new thread should run first (or
     * null if none). Workers are created with an initial first task
     * (in method execute()) to bypass queuing when there are fewer
     * than corePoolSize threads (in which case we always start one),
     * or when the queue is full (in which case we must bypass queue).
     * Initially idle threads are usually created via
     * prestartCoreThread or to replace other dying workers.
     *
     * @param core if true use corePoolSize as bound, else
     * maximumPoolSize. (A boolean indicator is used here rather than a
     * value to ensure reads of fresh values after checking other pool
     * state).
     * @return true if successful
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry: // 外層循環，負責判斷線程池狀態，處理線程池狀態變量加1操作
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c); // 讀取狀態值

            // Check if queue empty only if necessary.
            // 滿足下面兩大條件的，說明線程池不能接受任務了，直接返回false處理
            // 主要目的就是想說，只有線程池的狀態為 RUNNING 狀態時，線程池才會接收新的任務，增加新的Worker工作線程
            if (rs >= SHUTDOWN && // 線程池的狀態已經至少已經處于不能接收任務的狀態了，目的是檢查線程池是否處于關閉狀態
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            // 內層循環，負責worker數量加1操作
            for (;;) {
                int wc = workerCountOf(c); // 獲取當前worker線程數量
                if (wc >= CAPACITY || // 如果線程池數量達到最大上限值CAPACITY
                    // core為true時判斷是否大于corePoolSize核心線程數量
                    // core為false時判斷是否大于maximumPoolSize最大設置的線程數量
                    wc >= (core ? corePoolSize : maximumPoolSize)) 
                    return false;
                    
                // 調用CAS原子操作，目的是worker線程數量加1
                if (compareAndIncrementWorkerCount(c)) // 
                    break retry;
                    
                c = ctl.get();  // Re-read ctl // CAS原子操作失敗的話，則再次讀取ctl值
                if (runStateOf(c) != rs) // 如果剛剛讀取的c狀態不等于先前讀取的rs狀態，則繼續外層循環判斷
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
                // 之所以會CAS操作失敗，主要是由于多線程并發操作，導致workerCount工作線程數量改變而導致的，因此繼續內層循環嘗試操作
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            // 創建一個Worker工作線程對象，將任務firstTask，新創建的線程thread都封裝到了Worker對象里面
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                // 由于對工作線程集合workers的添加或者刪除，涉及到線程安全問題，所以才加上鎖且該鎖為非公平鎖
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    // 獲取鎖成功后，執行臨界區代碼，首先檢查獲取當前線程池的狀態rs
                    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); // 添加新的工作線程到工作線程集合workers，workers是set集合
                        int s = workers.size();
                        if (s > largestPoolSize) // 變量記錄了線程池在整個生命周期中曾經出現的最大線程個數
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) { // 往workers工作線程集合中添加成功后，則立馬調用線程start方法啟動起來
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted) // 如果啟動線程失敗的話，還得將剛剛添加成功的線程共集合中移除并且做線程數量做減1操作
                addWorkerFailed(w);
        }
        return workerStarted;
    }
    
2、小結：
    ? 該方法是任務提交的一個核心方法，主要完成狀態的檢查，工作線程的創建并添加到線程集合切最后順利的話將創建的線程啟動；
    ? addWorker(command, true)：當線程數小于corePoolSize時，添加一個需要處理的任務command進線程集合，如果workers數量超過corePoolSize時，則返回false不需要添加工作線程；
    ? addWorker(command, false)：當等待隊列已滿時，將新來的任務command添加到workers線程集合中去，若線程集合大小超過maximumPoolSize時，則返回false不需要添加工作線程；
    ? addWorker(null, false)：放一個空的任務進線程集合，當這個空任務的線程執行時，會從等待任務隊列中通過getTask獲取任務再執行，創建新線程且沒有任務分配，當執行時才去取任務；
    ? addWorker(null, true)：創建空任務的工作線程到workers集合中去，在setCorePoolSize方法調用時目的是初始化核心工作線程實例；

1、runWorker源碼：
    /**
     * Main worker run loop.  Repeatedly gets tasks from queue and
     * executes them, while coping with a number of issues:
     *
     * 1. We may start out with an initial task, in which case we
     * don"t need to get the first one. Otherwise, as long as pool is
     * running, we get tasks from getTask. If it returns null then the
     * worker exits due to changed pool state or configuration
     * parameters.  Other exits result from exception throws in
     * external code, in which case completedAbruptly holds, which
     * usually leads processWorkerExit to replace this thread.
     *
     * 2. Before running any task, the lock is acquired to prevent
     * other pool interrupts while the task is executing, and then we
     * ensure that unless pool is stopping, this thread does not have
     * its interrupt set.
     *
     * 3. Each task run is preceded by a call to beforeExecute, which
     * might throw an exception, in which case we cause thread to die
     * (breaking loop with completedAbruptly true) without processing
     * the task.
     *
     * 4. Assuming beforeExecute completes normally, we run the task,
     * gathering any of its thrown exceptions to send to afterExecute.
     * We separately handle RuntimeException, Error (both of which the
     * specs guarantee that we trap) and arbitrary Throwables.
     * Because we cannot rethrow Throwables within Runnable.run, we
     * wrap them within Errors on the way out (to the thread"s
     * UncaughtExceptionHandler).  Any thrown exception also
     * conservatively causes thread to die.
     *
     * 5. After task.run completes, we call afterExecute, which may
     * also throw an exception, which will also cause thread to
     * die. According to JLS Sec 14.20, this exception is the one that
     * will be in effect even if task.run throws.
     *
     * The net effect of the exception mechanics is that afterExecute
     * and the thread"s UncaughtExceptionHandler have as accurate
     * information as we can provide about any problems encountered by
     * user code.
     *
     * @param w the worker
     */
    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts 允許中斷
        boolean completedAbruptly = true;
        try {
            // 不斷從等待隊列blockingQueue中獲取任務
            // 之前addWorker(null, false)這樣的線程執行時，會通過getTask中再次獲取任務并執行
            while (task != null || (task = getTask()) != null) {
                w.lock(); // 上鎖，并不是防止并發執行任務，而是為了防止shutdown()被調用時不終止正在運行的worker線程
                // 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 {
                    // task.run()執行前，由子類實現
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run(); // 執行線程Runable的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 {
                        // task.run()執行后，由子類實現
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

2、小結：
    ? addWorker通過調用t.start()啟動了線程，線程池的真正核心執行任務的地方就在此runWorker中；
    ? 不斷的執行我們提交任務的run方法，可能是剛剛提交的任務，可能是隊列中等待的隊列，原因在于Worker工作線程類繼承了AQS類；
    ? Worker重寫了AQS的tryAcquire方法，不管先來后到，一種非公平的競爭機制，通過CAS獲取鎖，獲取到了就執行代碼塊，沒獲取到的話則添加到CLH隊列中通過利用LockSuporrt的park/unpark阻塞任務等待；
    ? addWorker通過調用t.start()啟動了線程，線程池的真正核心執行任務的地方就在此runWorker中；

1、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) {
        // 如果突然中止，說明runWorker中遇到什么異常了，那么正在工作的線程自然就需要減1操作了
        if (completedAbruptly) // If abrupt, then workerCount wasn"t adjusted
            decrementWorkerCount();

        final ReentrantLock mainLock = this.mainLock;
        // 執行到此，說明runWorker正常執行完了，需要正常退出工作線程，上鎖正常操作移除線程
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks; // 增加線程池完成任務數
            workers.remove(w); // 從workers線程集合中移除已經工作完的線程
        } finally {
            mainLock.unlock();
        }

        // 在對線程池有負效益的操作時，都需要“嘗試終止”線程池，主要是判斷線程池是否滿足終止的狀態；
        // 如果狀態滿足，但還有線程池還有線程，嘗試對其發出中斷響應，使其能進入退出流程；
        // 沒有線程了，更新狀態為tidying->terminated；
        tryTerminate();

        int c = ctl.get();
        
        // 如果狀態是running、shutdown，即tryTerminate()沒有成功終止線程池，嘗試再添加一個worker
        if (runStateLessThan(c, STOP)) {
            // 不是突然完成的，即沒有task任務可以獲取而完成的，計算min，并根據當前worker數量判斷是否需要addWorker()
            if (!completedAbruptly) {
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                
                // 如果min為0，且workQueue不為空，至少保持一個線程
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;
                    
                // 如果線程數量大于最少數量，直接返回，否則下面至少要addWorker一個
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            
            // 只要worker是completedAbruptly突然終止的，或者線程數量小于要維護的數量，就新添一個worker線程，即使是shutdown狀態
            addWorker(null, false);
        }
    }

2、小結：
    ? 異常中止情況worker數量減1，正常情況就上鎖從workers中移除；
    ? tryTerminate()：在對線程池有負效益的操作時，都需要“嘗試終止”線程池;
    ? 是否需要增加worker線程，如果線程池還沒有完全終止，仍需要保持一定數量的線程;    

1、如果是CPU密集型任務，就需要盡量壓榨CPU，參考值可以設為 NCPU+1；
2、如果是IO密集型任務，參考值可以設置為2*NCPU；

“強烈建議程序員使用較為方便的Executors工廠方法：

https://gitee.com/ylimhhmily/SpringCloudTutorial.git

SpringCloudTutorial交流QQ群: 235322432

SpringCloudTutorial交流微信群: 微信溝通群二維碼圖片鏈接

歡迎關注，您的肯定是對我最大的支持!!!