Java線程池架構(一)原理和源碼解析

在前面介紹JUC的文章中，提到了關于線程池Execotors的建立介紹，在文章：《

java之JUC系列-外部Tools

》中第一部分有詳細的說明，請參閱；

文章中其實說明了外部的使用方式，但是沒有說内部是如何實作的，為了加深對實作的了解，在使用中可以放心，我們這裡将做源碼解析以及回報到原理上，Executors工具可以建立普通的線程池以及schedule排程任務的排程池，其實兩者實作上還是有一些差別，但是了解了ThreadPoolExecutor，在看ScheduledThreadPoolExecutor就非常輕松了，後面的文章中也會專門介紹這塊，但是需要先看這篇文章。

使用Executors最常用的莫過于是使用：Executors.newFixedThreadPool(int)這個方法，因為它既可以限制數量，而且線程用完後不會一直被cache住；那麼就通過它來看看源碼，回過頭來再看其他構造方法的差別：

在《

》文章中提到了構造方法，為了和本文對接，再貼下代碼。

public static ExecutorService <strong>newFixedThreadPool</strong>(int nThreads) {

        return new ThreadPoolExecutor(nThreads, nThreads,

                                      0L, TimeUnit.MILLISECONDS,

                                      new LinkedBlockingQueue());

}

​

其實你可以自己new一個ThreadPoolExecutor，來達到自己的參數可控的程度，例如，可以将LinkedBlockingQueue換成其它的（如：SynchronousQueue），隻是可讀性會降低，這裡隻是使用了一種設計模式。

我們現在來看看ThreadPoolExecutor的源碼是怎麼樣的，也許你剛開始看他的源碼會很痛苦，因為你不知道作者為什麼是這樣設計的，是以本文就我看到的思想會給你做一個介紹，此時也許你通過知道了一些作者的思想，你也許就知道應該該如何去操作了。

這裡來看下構造方法中對那些屬性做了指派：

源碼段1：

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;

     }

這裡你可以看到最終指派的過程，可以先大概知道下參數的意思：

corePoolSize：核心運作的poolSize，也就是當超過這個範圍的時候，就需要将新的Runnable放入到等待隊列workQueue中了，我們把這些Runnable就叫做要去執行的任務吧。

maximumPoolSize：一般你用不到，當大于了這個值就會将任務由一個丢棄處理機制來處理，但是當你發生：newFixedThreadPool的時候，corePoolSize和maximumPoolSize是一樣的，而corePoolSize是先執行的，是以他會先被放入等待隊列，而不會執行到下面的丢棄進行中，看了後面的代碼你就知道了。

workQueue：等待隊列，當達到corePoolSize的時候，就向該等待隊列放入線程資訊（預設為一個LinkedBlockingQueue），運作中的線程屬性為：workers，為一個HashSet；我們的Runnable内部被包裝了一層，後面會看到這部分代碼；這個隊列預設是一個無界隊列（你也可以設定一個有界隊列），是以在生産者瘋狂生産的時候，考慮如何控制的問題。

keepAliveTime：預設都是0，當線程沒有任務處理後，保持多長時間，當你使用：newCachedThreadPool()，它将是60s的時間。這個參數在運作中的線程從workQueue擷取任務時，當(poolSize >corePoolSize || allowCoreThreadTimeOut)會用到，當然allowCoreThreadTimeOut要設定為true，也會先判定keepAliveTime是大于0的，不過由于它在corePoolSize上采用了Integer.MAX_VALUE，當遇到系統遇到瞬間沖擊，workers就會迅速膨脹，是以這個地方就不要去設定allowCoreThreadTimeOut=true，否則結果是這些運作中的線程會持續60s以上；另外，如果corePoolSize的值還沒到Integer.MAX_VALUE，當超過那個值以後，這些運作中的線程，也是

threadFactory：是構造Thread的方法，你可以自己去包裝和傳遞，主要實作newThread方法即可；

handler：也就是參數maximumPoolSize達到後丢棄處理的方法，java提供了5種丢棄處理的方法，當然你也可以自己根據實際情況去重寫，主要是要實作接口：RejectedExecutionHandler中的方法： public void rejectedExecution(Runnabler, ThreadPoolExecutor e) java預設的是使用：AbortPolicy，他的作用是當出現這中情況的時候會抛出一個異常；

其餘的還包含：

1、CallerRunsPolicy：如果發現線程池還在運作，就直接運作這個線程

2、DiscardOldestPolicy：線上程池的等待隊列中，将頭取出一個抛棄，然後将目前線程放進去。

3、DiscardPolicy：什麼也不做

4、AbortPolicy：java預設，抛出一個異常：RejectedExecutionException。

你可以自己寫一個，例如我們想在這個進行中，既不是完全丢棄，也不是完全啟動，也不是抛異常，而是控制生産者的線程，那麼你就可以嘗試某種方式将生産者的線程blocking住，其實就有點類似提到的Semaphor的功能了。

通常你得到線程池後，會調用其中的：submit方法或execute方法去操作；其實你會發現，submit方法最終會調用execute方法來進行操作，隻是他提供了一個Future來托管傳回值的處理而已，當你調用需要有傳回值的資訊時，你用它來處理是比較好的；這個Future會包裝對Callable資訊，并定義一個Sync對象（），當你發生讀取傳回值的操作的時候，會通過Sync對象進入鎖，直到有傳回值的資料通知，具體細節先不要看太多。

繼續向下，來看看execute最為核心的方法吧: 源碼段2：

     public void execute(Runnable command) {

         if (command == null)

         if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {

            if (runState == RUNNING && workQueue.offer(command)) {

                if (runState != RUNNING || poolSize == 0)

                    ensureQueuedTaskHandled(command);

            }

            else if (!addIfUnderMaximumPoolSize(command))

                reject(command); // is shutdown or saturated

        }

    }

這段代碼看似簡單，其實有點難懂，很多人也是這裡沒看懂，沒事，我一個if一個if說：

首先第一個判定空操作就不用說了，下面判定的poolSize >= corePoolSize成立時候會進入if的區域，當然它不成立也有可能會進入，他會判定addIfUnderCorePoolSize是否傳回false，如果傳回false就會進去；

我們先來看下addIfUnderCorePoolSize方法的源碼是什麼：

源碼段3：

private boolean addIfUnderCorePoolSize(Runnable firstTask) {

        Thread t = null;

        final ReentrantLock mainLock = this.mainLock;

        mainLock.lock();

        try {

            if (poolSize < corePoolSize && runState == RUNNING)

                t = addThread(firstTask);

        } finally {

            mainLock.unlock();

        if (t == null)

            return false;

        t.start();

        return true;

可以發現，這段源碼是如果發現小雨corePoolSize就會建立一個新的線程，并且調用線程的start()方法将線程運作起來:這個addThread()方法，我們先不考慮細節，因為我們還要先看到前面是怎麼進去的，這裡可以發信啊，隻有沒有建立成功Thread才會傳回false，也就是當目前的poolSize > corePoolSize的時候，或線程池已經不是在running狀态的時候才會出現；

注意：這裡在外部判定一次poolSize和corePoolSize隻是初步判定，内部是加鎖後判定的，以得到更為準确的結果，而外部初步判定如果是大于了，就沒有必要進入這段有鎖的代碼了。

此時我們知道了，目前線程數量大于corePoolSize的時候，就會進入【代碼段2】的第一個if語句中，回到【源碼段2】，繼續看if語句中的内容:

這裡标記為

源碼段4：

if (runState == RUNNING && workQueue.offer(command)) {

第一個if，也就是當目前狀态為running的時候，就會去執行workQueue.offer(command)，這個workQueue其實就是一個BlockingQueue，offer()操作就是在隊列的尾部寫入一個對象，此時寫入的對象為線程的對象而已；是以你可以認為隻有線程池在RUNNING狀态，才會在隊列尾部插入資料，否則就執行else if，其實else if可以看出是要做一個是否大于MaximumPoolSize的判定，如果大于這個值，就會做reject的操作，關于reject的說明，我們在【源碼段1】的解釋中已經非常明确的說明，這裡可以簡單看下源碼，以應征結果：

源碼段5：

    private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {

            if (poolSize < maximumPoolSize && runState == RUNNING)                 //在corePoolSize = maximumPoolSize下，該代碼幾乎不可能運作                 t = addThread(firstTask);         } finally {             mainLock.unlock();         }         if (t == null)             return false;         t.start();         return true; } void reject(Runnable command) {         handler.rejectedExecution(command, this);     } 

也就是如果線程池滿了，而且線程池調用了shutdown後，還在調用execute方法時，就會抛出上面說明的異常：RejectedExecutionException 再回頭來看下【代碼段4】中進入到等待隊列後的操作：

 if (runState != RUNNING || poolSize == 0)     ensureQueuedTaskHandled(command);

這段代碼是要線上程池運作狀态不是RUNNING或poolSize == 0才會調用，他是幹啥呢？ 他為什麼會不等于RUNNING呢？外面那一層不是判定了他== RUNNING了麼，其實有時間差就是了，如果是poolSize == 0也會執行這段代碼，但是裡面的判定條件是如果不是RUNNING，就做reject操作，在第一個線程進去的時候，會将第一個線程直接啟動起來；很多人也是看這段代碼很繞，因為不斷的循環判定類似的判定條件，你主要記住他們之間有時間差，要取最新的就好了。 此時貌似代碼看完了？咦，此時有問題了： 1、 等待中的線程在後來是如何跑起來的呢？線程池是不是有類似Timer一樣的守護程序不斷掃描線程隊列和等待隊列？還是利用某種鎖機制，實作類似wait和notify實作的？ 2、 線程池的運作隊列和等待隊列是如何管理的呢？這裡還沒看出影子呢！ NO，NO，NO！ Java在實作這部分的時候，使用了怪異的手段，神馬手段呢，還要再看一部分代碼才曉得。 在前面【源碼段3】中，我們看到了一個方法叫：addThread()，也許很少有人會想到關鍵在這裡，其實關鍵就是在這裡： 我們看看addThread()方法到底做了什麼。 源碼段6：

private Thread addThread(Runnable firstTask) {         Worker w = new Worker(firstTask);         Thread t = threadFactory.newThread(w);         if (t != null) {             w.thread = t;             workers.add(w);             int nt = ++poolSize;             if (nt > largestPoolSize)

                largestPoolSize = nt;

        return t;

這裡建立了一個Worker，其餘的操作，就是将poolSize++的操作，然後将将其放入workers的運作的HashSet中等操作；

我們主要關心Worker是幹什麼的，因為這個threadFactory對我們用途不大，隻是做了Thread的命名處理；而Worker你會發現它的定義也是一個Runnable，外部開始在代碼段中發現了調用哪個這個Worker的start()方法，也就是線程的啟動方法，其實也就是調用了Worker的run()方法，那麼我們重點要關心run方法是如何處理的

源碼段7：

public void run() {

            try {

                Runnable task = firstTask;

                firstTask = null;

                while (task != null || (task = getTask()) != null) {

                    runTask(task);

                    task = null;

                }

            } finally {

                workerDone(this);

FirstTask其實就是開始在建立work的時候，由外部傳入的Runnable對象，也就是你自己的Thread，你會發現它如果發現task為空，就會調用getTask()方法再判定，直到兩者為空，并且是一個while循環體。

那麼看看getTask()方法的實作為：

源碼段8：

     Runnable getTask() {

        for (;;) {

                int state = runState;

                if (state > SHUTDOWN)

                    return null;

                Runnable r;

                if (state == SHUTDOWN)  // Help drain queue

                    r = workQueue.poll();

                else if (poolSize > corePoolSize || allowCoreThreadTimeOut)

                    r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);

                else

                    r = workQueue.take();

                if (r != null)

                    return r;

                if (workerCanExit()) {

                    if (runState >= SHUTDOWN) // Wake up others

                        interruptIdleWorkers();

                // Else retry

            } catch (InterruptedException ie) {

                // On interruption, re-check runState

你會發現它是從workQueue隊列中，也就是等待隊列中擷取一個元素出來并傳回！

回過頭來根據代碼段6了解下：

目前線程運作完後，在到workQueue中去擷取一個task出來，繼續運作，這樣就保證了線程池中有一定的線程一直在運作；此時若跳出了while循環，隻有workQueue隊列為空才會出現或出現了類似于shutdown的操作，自然運作隊列會減少1，當再有新的線程進來的時候，就又開始向worker裡面放資料了，這樣以此類推，實作了線程池的功能。

這裡可以看下run方法的finally中調用的workerDone方法為：

源碼段9：

void workerDone(Worker w) {

            completedTaskCount += w.completedTasks;

            workers.remove(w);

            if (--poolSize == 0)

                tryTerminate();

注意這裡将workers.remove(w)掉，并且調用了—poolSize來做操作。

至于tryTerminate是做了更多關于回收方面的操作。

最後我們還要看一段代碼就是在【源碼段6】中出現的代碼調用為：runTask(task);這個方法也是運作的關鍵。

源碼段10：

private void runTask(Runnable task) {

            final ReentrantLock runLock = this.runLock;

            runLock.lock();

                if (runState < STOP &&                     Thread.interrupted() &&                     runState >= STOP)

                    thread.interrupt();

                boolean ran = false;

                beforeExecute(thread, task);

                try {

                    task.run();

                    ran = true;

                    afterExecute(task, null);

                    ++completedTasks;

                } catch (RuntimeException ex) {

                    if (!ran)

                        afterExecute(task, ex);

                    throw ex;

                runLock.unlock();

你可以看到，這裡面的task為傳入的task資訊，調用的不是start方法，而是run方法，因為run方法直接調用不會啟動新的線程，也是因為這樣，導緻了你無法擷取到你自己的線程的狀态，因為線程池是直接調用的run方法，而不是start方法來運作。

這裡有個beforeExecute和afterExecute方法，分别代表在執行前和執行後，你可以做一段操作，在這個類中，這兩個方法都是【空body】的，因為普通線程池無需做更多的操作。

如果你要實作類似暫停等待通知的或其他的操作，可以自己extends後進行重寫構造；

本文沒有介紹關于ScheduledThreadPoolExecutor調用的細節，下一篇文章會詳細說明，因為大部分代碼和本文一緻，差別在于一些細節，在介紹：ScheduledThreadPoolExecutor的時候，會明确的介紹它與Timer和TimerTask的巨大差別，差別不在于使用，而是在于本身内在的處理細節。