JDK核心源碼之FixedThreadPool源碼註釋

    今天，咱們來分析一下併發裏面關於線程池的原理，咱們經過源碼深刻剖析，這樣瞭解到其中的細節，對於咱們把握好併發的線程池使用很是有幫助。

    你們都知道線程池的類型有好幾個，可是咱們只要搞清楚其中的一個，其他的其實很好分析。這兒就以FixedThreadPool舉例。從源碼的角度去理解這個線程池。

首先先建立一個線程池//建立線程池ExecutorService executorService = Executors.newFixedThreadPool(4);        executorService.execute(new Runnable() {            public void run() {                System.out.println("test");            }        });跟蹤newFixedThreadPool方法，代碼以下：    public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());​}

 

很顯然咱們看到這個是重要的方法是類ThreadPoolExecutor的，點進去：

 

//corePoolSize:核心線程數//maximumPoolSize:線程池裏最多能夠建立多少線程//keepAliveTime:線程能夠存活多久//TimeUnit: 時間的單位，配合keepAliveTime一塊兒使用//workQueue: 若是線程的個數超過核心線程數的數量了，就會把多出來的線程放入到這個隊列，若是這個隊列也存滿了，而後就會在建立出線程，可是總的線程不能多於maximumPoolSize的值。public ThreadPoolExecutor(int corePoolSize,                          int maximumPoolSize,                          long keepAliveTime,                          TimeUnit unit,                          BlockingQueue<Runnable> workQueue) {        //給了一個默認的線程工廠        //給了默認的拒絕策略        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             Executors.defaultThreadFactory(), defaultHandler)    }點擊去看構造函數：以下的構造函數，其實也沒什麼特別的，就是通常的構造函數。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;​}

 

都分析到構造函數的位置了，那麼按照咱們通常分析源碼的習慣，確定須要看一下，這個類裏面主要的成員變量和核心方法。以下變量須要咱們注意：

//這個變量很是的重要，//高3位表明這線程的狀態//低29位表明着線程的數量private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//29private static final int COUNT_BITS = Integer.SIZE - 3;// 1 << 29 - 1  = 00011111 11111111 11111111 11111111//1^29 - 1;private static final int CAPACITY   = (1 << COUNT_BITS) - 1;// 線程池的狀態// 11100000 00000000 00000000 00000000private static final int RUNNING    = -1 << COUNT_BITS;//獲取線程的狀態private static int runStateOf(int c)     {return c & ~CAPACITY;}//獲取線程的個數private static int workerCountOf(int c)  {return c & CAPACITY;}private static int ctlOf(int rs, int wc) {return rs | wc;   }​​ 

如上，咱們已經分析完了線程池的初始化工做。接下來咱們分析一下execute()方法。

ExecutorService executorService = Executors.newFixedThreadPool(10);//接下來咱們就要分析這個方法//這個是抽象方法，咱們看其實現類//ThreadPoolException裏面的excute方法//這個方法很是重要executorService.execute(new Runnable() {public void run() {System.out.println("test");}});

 

 

  進到ThreadPoolException之後，咱們在裏面找到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.           1. 若是正在運行的線程數小於核心線程數，那麼就嘗試啓動一個新的線程。              經過調用addWorker方法建立新的線程，建立線程的時候須要檢測一下線程的狀態              和線程總數         * 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.           2.若是發現正在運行的線程的個數已經大於corePoolSize,那麼就把當前的線程加入到           隊列中。         * 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.           3. 若是隊列裏面也放不下了，那麼就會啓動新的線程，可是總的線程的個數不能大於            maximumPoolSize         */        int c = ctl.get();        //第一步：若是當前線程的個數 小於 核心線程數        if (workerCountOf(c) < corePoolSize) {            //經過addWorker的方式啓動線程            //true表明的是建立的是核心線程            if (addWorker(command, true))                return;            c = ctl.get();        }        //第二步：若是代碼走到這兒，說明當前線程的個數已經大於corePoolSize了        //就把線程加入到隊列裏面去        //注意其實offer這個方法是不阻塞的        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);        }        //第三步：代碼走到這兒說明隊列存滿了        //要建立新的線程，可是總的線程不能大於maximumPoolSize        //false表明的是建立非核心線程。        else if (!addWorker(command, false))            reject(command);}​​

     上面的核心代碼畫成圖以下：

 

   咱們看了execute()方法的三個步驟之後，咱們接下來一個步驟一個步驟的去分析，你們注意到第一步和第三步 建立線程的時候都調用的是addWorker的方法，不一樣的是傳進去的參數一個是true，一個是false，true表明建立的是核心線程，false表明建立的是非核心線程。

   首先分析第一步的建立核心線程的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.            //這兒能夠不用多管就是檢查一下            if (rs >= SHUTDOWN &&                ! (rs == SHUTDOWN &&                   firstTask == null &&                   ! workQueue.isEmpty()))                return false;            for (;;) {                //獲取當前的線程數                int wc = workerCountOf(c);                //若是當前線程數大於總的線程數或者                //當前的線程大於corePoolSize 或者大於maximumPoolSize               //都返回false                //當前core是ture 因此該判斷是： wc >= corePoolSize                //因此是若是當前線程數大於設置的核心線程數九返回false                if (wc >= CAPACITY ||                    wc >= (core ? corePoolSize : maximumPoolSize))                    return false;                //增長當前的線程數                //增長成功就跳出當前循環                //使用CAS的方式增長了線程數，保證了線程安全                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) {                //這兒加了鎖                //這兒爲何加鎖，其實咱們想一下就明白了                // 咱們建立了一個線程對象（worker）,而後要把這個worker                // 放入到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.                    //獲取線程池的狀態                    int rs = runStateOf(ctl.get());                    //若是狀態正常                    if (rs < SHUTDOWN ||                        (rs == SHUTDOWN && firstTask == null)) {                        if (t.isAlive()) // precheck that t is startable                            throw new IllegalThreadStateException();                        //把當前封裝的線程加入到workers裏面                        workers.add(w);                        int s = workers.size();                        if (s > largestPoolSize)                            largestPoolSize = s;                        workerAdded = true;                    }                } finally {                    //釋放鎖                    mainLock.unlock();                }                //若是正常執行的話，代碼會走到這兒                if (workerAdded) {                    //調用t的start方法                    //這個方法較爲重要，接下來咱們就是分析這個方法                   //這兒值得咱們注意的是這個t是Worker裏面的t                   //因此咱們要想知道這個t.start()最終調的什麼方法                   //咱們須要去看一下worker的內部                    t.start();                   workerStarted = true;                }            }        } finally {            if (! workerStarted)                addWorkerFailed(w);        }        return workerStarted;    }​

  接下來咱們分析一下t.start方法

  可是咱們從以前的代碼得知，咱們是經過這個以下代碼獲取到的t:

  

//fristTask就是咱們傳進來的代碼w = new Worker(firstTask);final Thread t = w.thread;

   想知道t是啥，那麼咱們進入看一下Worker的構造函數：

   

Worker(Runnable firstTask) {   setState(-1); // inhibit interrupts until runWorker    this.firstTask = firstTask;    //這個實際上是使用默認的工程建立的一個線程   //this就是Worker本身，Worker自己是實現了Runnable接口，因此他本身   //也是一個線程。因此到這個時候咱們就應該明白了。   //前面的t.start方法，其實調用的就是worker的run()方法    this.thread = getThreadFactory().newThread(this);}

 原來t是一個worker默認的線程工廠建立的線程，這樣咱們就得知t.start其實調用的是Worker的run方法，接下來咱們分析一下，Worker的run方法。

  

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

 調用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 {        //一開始咱們建立線程的時候，這個時候task不等於null       //可是若是核心線程裏面已經把task運行完了，那麼這個時候task就爲null       //若是task爲null了，那麼就會走getTask這個方法      //這個方法就會從無界隊列裏面獲取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            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;                //統計信息                w.completedTasks++;                w.unlock();            }        }        completedAbruptly = false;    } finally {         //異常處理        processWorkerExit(w, completedAbruptly);    }}

到此爲止，咱們分析到了前面說的三個不步驟的第一步，使用核心線程啓動任務（咱們後面才分析啓動任務之後，代碼是如何運行的？）。接下來咱們分析一下第二步，如何把在覈心線程滿了之後的新任務壓入隊列。咱們回過頭來看這個方法：

ThreadPoolExcutor的execute方法，貼出第二步的關鍵代碼：

//若是線程池的狀態是正在運行，那麼就把咱們寫的線程壓入到 //orkerQueue裏面，由於咱們分析的是FixedThreadPool線程池 //因此這個workerQueue就是LinkedBlockingQueue,並且這個隊列默認的大小給了一個Integer的最大值。 //接下來咱們分析一下這個offer方法，不過這兒值得咱們注意的是 //雖然咱們知道LinkedBlockingQueue是一個阻塞隊列，可是其offer方法是不阻塞的。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);}

 咱們能夠進去看一下offer方法：

public boolean offer(E e) {    if (e == null) throw new NullPointerException();    final AtomicInteger count = this.count;    //重要，這個的意思就是，若是隊列滿了就返回false    //可是咱們的這個隊列很大，咱們能夠暫時認爲通常狀況下是不會滿的。    if (count.get() == capacity)        return false;    int c = -1;    Node<E> node = new Node<E>(e);    final ReentrantLock putLock = this.putLock;    putLock.lock();    try {         //若是隊列裏面的線程數小於容量        if (count.get() < capacity) {            //壓入隊列            enqueue(node);            //當前存入的任務個數增長1            c = count.getAndIncrement();            //若是當前容量還能存數據            if (c + 1 < capacity)            //喚醒線程，其實咱們以前說過offer是沒有阻塞功能的           //這兒的singnal主要是是配合take去使用，take方法有阻塞功能。                notFull.signal();        }    } finally {        putLock.unlock();    }    if (c == 0)        signalNotEmpty();    return c >= 0;}

    那接下來咱們分析ThreadPoolExcutor的execute方法裏面的第三步，貼出關鍵源碼以下：

//這方法咱們知道，由於前面的兩個條件都不知足，也就是，核心線程也佔用滿了//隊列也存滿了，因此代碼就會走到這兒。//這兒進去走的是addWorker的方法，其實這個方法咱們前面就分析過了，只不錯前面傳進去的//參數是true（表明是核心線程），如今的參數是false表明是非核心線程。else if (!addWorker(command, false))reject(command);

  

到目前爲止咱們分析完了三種策略：1）剛開始的線程先用核心線程處理 2）核心線程在忙，那麼就把當前線程壓到隊列 3）隊列滿了，那麼就啓動

非核心線程。不過須要咱們注意的是咱們是以FixedThreadPool線程池爲例的，FixedThreadPool使用的隊列是LinkedBlockingQueue,這個默認給的是

是int的最大值，換句話說咱們能夠認爲就是無界的，由於在當要把這個隊列存滿的時候內存早已經溢出了，也就是說，咱們的第三個策略壓根就走不到。

前面咱們分析了第一種策略的時候，咱們順帶也分析了，核心線程獲取到須要執行的task之後是如何執行task的，接下來咱們須要分析。核心線程把當前任務已經處理完了，是如何從隊列裏面獲取task任務進行處理的。

入口是從addWorker這個方法進去，以前咱們這個方法只是分析到啓動了線程，可是沒有分析裏面的線程是如何工做的，接下來分析以前啓動的線程是如何工做的。

以下代碼咱們以前看到過：

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) {            //核心代碼           //咱們以前分析過，其實這個start方法調用的是worker的run方法。           //worker的run方法裏調用的是runWork方法           //咱們回過頭來再分析一下，runworker方法            t.start();            workerStarted = true;        }    }} finally {    if (! workerStarted)        addWorkerFailed(w);}​final void runWorker(Worker w) {    Thread wt = Thread.currentThread();    Runnable task = w.firstTask;    w.firstTask = null;    w.unlock(); // allow interrupts    boolean completedAbruptly = true;    try {        //核心代碼        //分析一下：若是task != null         //說明當前核心線程有任務運行，而後直接運行task便可。        //若是task == null 那麼說明核心線程已經把任務運行完了，         //那麼由於條件是|| 因此接下        //來會走task = getTask()) != null 代碼， getTask()就是從隊列裏面去獲取任務        //而後運行的。        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 {                    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);    }}

  接下來分析一下getTask方法，看一下是如何從隊列裏面獲取task任務的？

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.        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {            decrementWorkerCount();            return null;        }        int wc = workerCountOf(c);        // Are workers subject to culling?        //allowCoreThreadTimeOut 默認是false        //當前線程數有沒有超過corePoolSize 正常狀況下是flase        //因此timed的值是false        //注：allowCoreThreadTimeOut=true表明着，核心線程若是空閒超過必定時間就回收        //超過核心線程的線程 空閒超過必定時間後會回收線程。        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;        if ((wc > maximumPoolSize || (timed && timedOut))           && (wc > 1 || workQueue.isEmpty())) {            if (compareAndDecrementWorkerCount(c))                return null;            continue;        }        try {            //這個地方代碼比較關鍵            //若是timed = ture 那麼執行的是 workQueue.poll方法  這個方法沒有阻塞的功能            //若是timed = false那麼指定是 workerQueue.take方法 若是方法帶有阻塞的功能            Runnable r = timed ?                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                //執行這個方法，從阻塞隊列中獲取任務。                workQueue.take();            if (r != null)                return r;            timedOut = true;        } catch (InterruptedException retry) {           timedOut = false;        }    }}

 

   到此其實咱們把FixedThreadPool的原理就搞明白了，其他的幾個線程池比較簡單，你們能夠本身看一下。咱們畫圖總結一下這個線程池的原理：