Java并發程式設計：線程池的使用

以下是本文的目錄大綱：

　　一.java中的threadpoolexecutor類

　　二.深入剖析線程池實作原理

　　三.使用示例

　　四.如何合理配置線程池的大小　

　　若有不正之處請多多諒解，并歡迎批評指正。

　　請尊重作者勞動成果，轉載請标明原文連結：

一.java中的threadpoolexecutor類

　　java.uitl.concurrent.threadpoolexecutor類是線程池中最核心的一個類，是以如果要透徹地了解java中的線程池，必須先了解這個類。下面我們來看一下threadpoolexecutor類的具體實作源碼。

　　在threadpoolexecutor類中提供了四個構造方法：

public class threadpoolexecutor extends abstractexecutorservice {

}

　　從上面的代碼可以得知，threadpoolexecutor繼承了abstractexecutorservice類，并提供了四個構造器，事實上，通過觀察每個構造器的源碼具體實作，發現前面三個構造器都是調用的第四個構造器進行的初始化工作。

　　下面解釋下一下構造器中各個參數的含義：

corepoolsize：核心池的大小，這個參數跟後面講述的線程池的實作原理有非常大的關系。在建立了線程池後，預設情況下，線程池中并沒有任何線程，而是等待有任務到來才建立線程去執行任務，除非調用了prestartallcorethreads()或者prestartcorethread()方法，從這2個方法的名字就可以看出，是預建立線程的意思，即在沒有任務到來之前就建立corepoolsize個線程或者一個線程。預設情況下，在建立了線程池後，線程池中的線程數為0，當有任務來之後，就會建立一個線程去執行任務，當線程池中的線程數目達到corepoolsize後，就會把到達的任務放到緩存隊列當中；

maximumpoolsize：線程池最大線程數，這個參數也是一個非常重要的參數，它表示線上程池中最多能建立多少個線程；

keepalivetime：表示線程沒有任務執行時最多保持多久時間會終止。預設情況下，隻有當線程池中的線程數大于corepoolsize時，keepalivetime才會起作用，直到線程池中的線程數不大于corepoolsize，即當線程池中的線程數大于corepoolsize時，如果一個線程空閑的時間達到keepalivetime，則會終止，直到線程池中的線程數不超過corepoolsize。但是如果調用了allowcorethreadtimeout(boolean)方法，線上程池中的線程數不大于corepoolsize時，keepalivetime參數也會起作用，直到線程池中的線程數為0；

unit：參數keepalivetime的時間機關，有7種取值，在timeunit類中有7種靜态屬性：

複制代碼

timeunit.days; //天

timeunit.hours; //小時

timeunit.minutes; //分鐘

timeunit.seconds; //秒

timeunit.milliseconds; //毫秒

timeunit.microseconds; //微妙

timeunit.nanoseconds; //納秒

workqueue：一個阻塞隊列，用來存儲等待執行的任務，這個參數的選擇也很重要，會對線程池的運作過程産生重大影響，一般來說，這裡的阻塞隊列有以下幾種選擇：

arrayblockingqueue;

linkedblockingqueue;

synchronousqueue;

　　arrayblockingqueue和priorityblockingqueue使用較少，一般使用linkedblockingqueue和synchronous。線程池的排隊政策與blockingqueue有關。

threadfactory：線程工廠，主要用來建立線程；

handler：表示當拒絕處理任務時的政策，有以下四種取值：

threadpoolexecutor.abortpolicy:丢棄任務并抛出rejectedexecutionexception異常。

threadpoolexecutor.discardpolicy：也是丢棄任務，但是不抛出異常。

threadpoolexecutor.discardoldestpolicy：丢棄隊列最前面的任務，然後重新嘗試執行任務（重複此過程）

threadpoolexecutor.callerrunspolicy：由調用線程處理該任務

具體參數的配置與線程池的關系将在下一節講述。

從上面給出的threadpoolexecutor類的代碼可以知道，threadpoolexecutor繼承了abstractexecutorservice，我們來看一下abstractexecutorservice的實作：

public abstract class abstractexecutorservice implements executorservice {

　　abstractexecutorservice是一個抽象類，它實作了executorservice接口。

　　我們接着看executorservice接口的實作：

public interface executorservice extends executor {

　　而executorservice又是繼承了executor接口，我們看一下executor接口的實作：

public interface executor {

　　到這裡，大家應該明白了threadpoolexecutor、abstractexecutorservice、executorservice和executor幾個之間的關系了。

　　executor是一個頂層接口，在它裡面隻聲明了一個方法execute(runnable)，傳回值為void，參數為runnable類型，從字面意思可以了解，就是用來執行傳進去的任務的；

　　然後executorservice接口繼承了executor接口，并聲明了一些方法：submit、invokeall、invokeany以及shutdown等；

　　抽象類abstractexecutorservice實作了executorservice接口，基本實作了executorservice中聲明的所有方法；

　　然後threadpoolexecutor繼承了類abstractexecutorservice。

　　在threadpoolexecutor類中有幾個非常重要的方法：

execute()

submit()

shutdown()

shutdownnow()

　　execute()方法實際上是executor中聲明的方法，在threadpoolexecutor進行了具體的實作，這個方法是threadpoolexecutor的核心方法，通過這個方法可以向線程池送出一個任務，交由線程池去執行。

　　submit()方法是在executorservice中聲明的方法，在abstractexecutorservice就已經有了具體的實作，在threadpoolexecutor中并沒有對其進行重寫，這個方法也是用來向線程池送出任務的，但是它和execute()方法不同，它能夠傳回任務執行的結果，去看submit()方法的實作，會發現它實際上還是調用的execute()方法，隻不過它利用了future來擷取任務執行結果（future相關内容将在下一篇講述）。

　　shutdown()和shutdownnow()是用來關閉線程池的。

　　還有很多其他的方法：

　　比如：getqueue() 、getpoolsize() 、getactivecount()、getcompletedtaskcount()等擷取與線程池相關屬性的方法，有興趣的朋友可以自行查閱api。

二.深入剖析線程池實作原理

　　在上一節我們從宏觀上介紹了threadpoolexecutor，下面我們來深入解析一下線程池的具體實作原理，将從下面幾個方面講解：

　　1.線程池狀态

　　2.任務的執行

　　3.線程池中的線程初始化

　　4.任務緩存隊列及排隊政策

　　5.任務拒絕政策

　　6.線程池的關閉

　　7.線程池容量的動态調整

1.線程池狀态

在threadpoolexecutor中定義了一個volatile變量，另外定義了幾個static final變量表示線程池的各個狀态：

volatile int runstate;

static final int running = 0;

static final int shutdown = 1;

static final int stop = 2;

static final int terminated = 3;

　　runstate表示目前線程池的狀态，它是一個volatile變量用來保證線程之間的可見性；

　　下面的幾個static final變量表示runstate可能的幾個取值。

　　當建立線程池後，初始時，線程池處于running狀态；

　　如果調用了shutdown()方法，則線程池處于shutdown狀态，此時線程池不能夠接受新的任務，它會等待所有任務執行完畢；

　　如果調用了shutdownnow()方法，則線程池處于stop狀态，此時線程池不能接受新的任務，并且會去嘗試終止正在執行的任務；

　　當線程池處于shutdown或stop狀态，并且所有工作線程已經銷毀，任務緩存隊列已經清空或執行結束後，線程池被設定為terminated狀态。

2.任務的執行

　　在了解将任務送出給線程池到任務執行完畢整個過程之前，我們先來看一下threadpoolexecutor類中其他的一些比較重要成員變量：

private final blockingqueue workqueue; //任務緩存隊列，用來存放等待執行的任務

private final reentrantlock mainlock = new reentrantlock(); //線程池的主要狀态鎖，對線程池狀态（比如線程池大小

private final hashset workers = new hashset(); //用來存放工作集

private volatile long keepalivetime; //線程存貨時間

private volatile boolean allowcorethreadtimeout; //是否允許為核心線程設定存活時間

private volatile int corepoolsize; //核心池的大小（即線程池中的線程數目大于這個參數時，送出的任務會被放進任務緩存隊列）

private volatile int maximumpoolsize; //線程池最大能容忍的線程數

private volatile int poolsize; //線程池中目前的線程數

private volatile rejectedexecutionhandler handler; //任務拒絕政策

private volatile threadfactory threadfactory; //線程工廠，用來建立線程

private int largestpoolsize; //用來記錄線程池中曾經出現過的最大線程數

private long completedtaskcount; //用來記錄已經執行完畢的任務個數

　　每個變量的作用都已經标明出來了，這裡要重點解釋一下corepoolsize、maximumpoolsize、largestpoolsize三個變量。

　　corepoolsize在很多地方被翻譯成核心池大小，其實我的了解這個就是線程池的大小。舉個簡單的例子：

　　假如有一個工廠，工廠裡面有10個勞工，每個勞工同時隻能做一件任務。

　　是以隻要當10個勞工中有勞工是空閑的，來了任務就配置設定給空閑的勞工做；

　　當10個勞工都有任務在做時，如果還來了任務，就把任務進行排隊等待；

　　如果說新任務數目增長的速度遠遠大于勞工做任務的速度，那麼此時工廠主管可能會想補救措施，比如重新招4個臨時勞工進來；

　　然後就将任務也配置設定給這4個臨時勞工做；

　　如果說着14個勞工做任務的速度還是不夠，此時工廠主管可能就要考慮不再接收新的任務或者抛棄前面的一些任務了。

　　當這14個勞工當中有人空閑時，而新任務增長的速度又比較緩慢，工廠主管可能就考慮辭掉4個臨時工了，隻保持原來的10個勞工，畢竟請額外的勞工是要花錢的。

　　這個例子中的corepoolsize就是10，而maximumpoolsize就是14（10+4）。

　　也就是說corepoolsize就是線程池大小，maximumpoolsize在我看來是線程池的一種補救措施，即任務量突然過大時的一種補救措施。

　　不過為了友善了解，在本文後面還是将corepoolsize翻譯成核心池大小。

　　largestpoolsize隻是一個用來起記錄作用的變量，用來記錄線程池中曾經有過的最大線程數目，跟線程池的容量沒有任何關系。

　　下面我們進入正題，看一下任務從送出到最終執行完畢經曆了哪些過程。

　　在threadpoolexecutor類中，最核心的任務送出方法是execute()方法，雖然通過submit也可以送出任務，但是實際上submit方法裡面最終調用的還是execute()方法，是以我們隻需要研究execute()方法的實作原理即可：

public void execute(runnable command) {

　　上面的代碼可能看起來不是那麼容易了解，下面我們一句一句解釋：

　　首先，判斷送出的任務command是否為null，若是null，則抛出空指針異常；

　　接着是這句，這句要好好了解一下：

if (poolsize >= corepoolsize || !addifundercorepoolsize(command))

　　由于是或條件運算符，是以先計算前半部分的值，如果線程池中目前線程數不小于核心池大小，那麼就會直接進入下面的if語句塊了。

　　如果線程池中目前線程數小于核心池大小，則接着執行後半部分，也就是執行

addifundercorepoolsize(command)

　　如果執行完addifundercorepoolsize這個方法傳回false，則繼續執行下面的if語句塊，否則整個方法就直接執行完畢了。

　　如果執行完addifundercorepoolsize這個方法傳回false，然後接着判斷：

if (runstate == running && workqueue.offer(command))

　　如果目前線程池處于running狀态，則将任務放入任務緩存隊列；如果目前線程池不處于running狀态或者任務放入緩存隊列失敗，則執行：

addifundermaximumpoolsize(command)

　　如果執行addifundermaximumpoolsize方法失敗，則執行reject()方法進行任務拒絕處理。

　　回到前面：

　　這句的執行，如果說目前線程池處于running狀态且将任務放入任務緩存隊列成功，則繼續進行判斷：

if (runstate != running || poolsize == 0)

　　這句判斷是為了防止在将此任務添加進任務緩存隊列的同時其他線程突然調用shutdown或者shutdownnow方法關閉了線程池的一種應急措施。如果是這樣就執行：

ensurequeuedtaskhandled(command)

　　進行應急處理，從名字可以看出是保證 添加到任務緩存隊列中的任務得到處理。

　　我們接着看2個關鍵方法的實作：addifundercorepoolsize和addifundermaximumpoolsize：

private boolean addifundercorepoolsize(runnable firsttask) {

　　這個是addifundercorepoolsize方法的具體實作，從名字可以看出它的意圖就是當低于核心吃大小時執行的方法。下面看其具體實作，首先擷取到鎖，因為這地方涉及到線程池狀态的變化，先通過if語句判斷目前線程池中的線程數目是否小于核心池大小，有朋友也許會有疑問：前面在execute()方法中不是已經判斷過了嗎，隻有線程池目前線程數目小于核心池大小才會執行addifundercorepoolsize方法的，為何這地方還要繼續判斷？原因很簡單，前面的判斷過程中并沒有加鎖，是以可能在execute方法判斷的時候poolsize小于corepoolsize，而判斷完之後，在其他線程中又向線程池送出了任務，就可能導緻poolsize不小于corepoolsize了，是以需要在這個地方繼續判斷。然後接着判斷線程池的狀态是否為running，原因也很簡單，因為有可能在其他線程中調用了shutdown或者shutdownnow方法。然後就是執行

1

t = addthread(firsttask);

　　這個方法也非常關鍵，傳進去的參數為送出的任務，傳回值為thread類型。然後接着在下面判斷t是否為空，為空則表明建立線程失敗（即poolsize>=corepoolsize或者runstate不等于running），否則調用t.start()方法啟動線程。

　　我們來看一下addthread方法的實作：

private thread addthread(runnable firsttask) {

在addthread方法中，首先用送出的任務建立了一個worker對象，然後調用線程工廠threadfactory建立了一個新的線程t，然後将線程t的引用指派給了worker對象的成員變量thread，接着通過workers.add(w)将worker對象添加到工作集當中。

下面我們看一下worker類的實作：

private final class worker implements runnable {

　　它實際上實作了runnable接口，是以上面的thread t = threadfactory.newthread(w);效果跟下面這句的效果基本一樣：

thread t = new thread(w);

相當于傳進去了一個runnable任務，線上程t中執行這個runnable。

既然worker實作了runnable接口，那麼自然最核心的方法便是run()方法了：

public void run() {

從run方法的實作可以看出，它首先執行的是通過構造器傳進來的任務firsttask，在調用runtask()執行完firsttask之後，在while循環裡面不斷通過gettask()去取新的任務來執行，那麼去哪裡取呢？自然是從任務緩存隊列裡面去取，gettask是threadpoolexecutor類中的方法，并不是worker類中的方法，下面是gettask方法的實作：

runnable gettask() {

　　在gettask中，先判斷目前線程池狀态，如果runstate大于shutdown（即為stop或者terminated），則直接傳回null。

　　如果runstate為shutdown或者running，則從任務緩存隊列取任務。

　　如果目前線程池的線程數大于核心池大小corepoolsize或者允許為核心池中的線程設定空閑存活時間，則調用poll(time,timeunit)來取任務，這個方法會等待一定的時間，如果取不到任務就傳回null。

　　然後判斷取到的任務r是否為null，為null則通過調用workercanexit()方法來判斷目前worker是否可以退出，我們看一下workercanexit()的實作：

private boolean workercanexit() {

也就是說如果線程池處于stop狀态、或者任務隊列已為空或者允許為核心池線程設定空閑存活時間并且線程數大于1時，允許worker退出。如果允許worker退出，則調用interruptidleworkers()中斷處于空閑狀态的worker，我們看一下interruptidleworkers()的實作：

void interruptidleworkers() {

從實作可以看出，它實際上調用的是worker的interruptifidle()方法，在worker的interruptifidle()方法中：

void interruptifidle() {

這裡有一個非常巧妙的設計方式，假如我們來設計線程池，可能會有一個任務分派線程，當發現有線程空閑時，就從任務緩存隊列中取一個任務交給空閑線程執行。但是在這裡，并沒有采用這樣的方式，因為這樣會要額外地對任務分派線程進行管理，無形地會增加難度和複雜度，這裡直接讓執行完任務的線程去任務緩存隊列裡面取任務來執行。

我們再看addifundermaximumpoolsize方法的實作，這個方法的實作思想和addifundercorepoolsize方法的實作思想非常相似，唯一的差別在于addifundermaximumpoolsize方法是線上程池中的線程數達到了核心池大小并且往任務隊列中添加任務失敗的情況下執行的：

private boolean addifundermaximumpoolsize(runnable firsttask) {

　　看到沒有，其實它和addifundercorepoolsize方法的實作基本一模一樣，隻是if語句判斷條件中的poolsize < maximumpoolsize不同而已。

　　到這裡，大部分朋友應該對任務送出給線程池之後到被執行的整個過程有了一個基本的了解，下面總結一下：

　　1）首先，要清楚corepoolsize和maximumpoolsize的含義；

　　2）其次，要知道worker是用來起到什麼作用的；

　　3）要知道任務送出給線程池之後的處理政策，這裡總結一下主要有4點：

如果目前線程池中的線程數目小于corepoolsize，則每來一個任務，就會建立一個線程去執行這個任務；

如果目前線程池中的線程數目>=corepoolsize，則每來一個任務，會嘗試将其添加到任務緩存隊列當中，若添加成功，則該任務會等待空閑線程将其取出去執行；若添加失敗（一般來說是任務緩存隊列已滿），則會嘗試建立新的線程去執行這個任務；

如果目前線程池中的線程數目達到maximumpoolsize，則會采取任務拒絕政策進行處理；

如果線程池中的線程數量大于 corepoolsize時，如果某線程空閑時間超過keepalivetime，線程将被終止，直至線程池中的線程數目不大于corepoolsize；如果允許為核心池中的線程設定存活時間，那麼核心池中的線程空閑時間超過keepalivetime，線程也會被終止。

3.線程池中的線程初始化

　　預設情況下，建立線程池之後，線程池中是沒有線程的，需要送出任務之後才會建立線程。

　　在實際中如果需要線程池建立之後立即建立線程，可以通過以下兩個方法辦到：

prestartcorethread()：初始化一個核心線程；

prestartallcorethreads()：初始化所有核心線程

下面是這2個方法的實作：

public boolean prestartcorethread() {

public int prestartallcorethreads() {

注意上面傳進去的參數是null，根據第2小節的分析可知如果傳進去的參數為null，則最後執行線程會阻塞在gettask方法中的

r = workqueue.take();

　　即等待任務隊列中有任務。

4.任務緩存隊列及排隊政策

　　在前面我們多次提到了任務緩存隊列，即workqueue，它用來存放等待執行的任務。

　　workqueue的類型為blockingqueue，通常可以取下面三種類型：

　　1）arrayblockingqueue：基于數組的先進先出隊列，此隊列建立時必須指定大小；

　　2）linkedblockingqueue：基于連結清單的先進先出隊列，如果建立時沒有指定此隊列大小，則預設為integer.max_value；

　　3）synchronousqueue：這個隊列比較特殊，它不會儲存送出的任務，而是将直接建立一個線程來執行新來的任務。

5.任務拒絕政策

當線程池的任務緩存隊列已滿并且線程池中的線程數目達到maximumpoolsize，如果還有任務到來就會采取任務拒絕政策，通常有以下四種政策：

6.線程池的關閉

　　threadpoolexecutor提供了兩個方法，用于線程池的關閉，分别是shutdown()和shutdownnow()，其中：

shutdown()：不會立即終止線程池，而是要等所有任務緩存隊列中的任務都執行完後才終止，但再也不會接受新的任務

shutdownnow()：立即終止線程池，并嘗試打斷正在執行的任務，并且清空任務緩存隊列，傳回尚未執行的任務

7.線程池容量的動态調整

　　threadpoolexecutor提供了動态調整線程池容量大小的方法：setcorepoolsize()和setmaximumpoolsize()，

setcorepoolsize：設定核心池大小

setmaximumpoolsize：設定線程池最大能建立的線程數目大小

　　當上述參數從小變大時，threadpoolexecutor進行線程指派，還可能立即建立新的線程來執行任務。

三.使用示例

前面我們讨論了關于線程池的實作原理，這一節我們來看一下它的具體使用：

public class test {

class mytask implements runnable {

　　執行結果：

正在執行task 0

線程池中線程數目：1，隊列中等待執行的任務數目：0，已執行玩别的任務數目：0

線程池中線程數目：2，隊列中等待執行的任務數目：0，已執行玩别的任務數目：0

正在執行task 1

線程池中線程數目：3，隊列中等待執行的任務數目：0，已執行玩别的任務數目：0

正在執行task 2

線程池中線程數目：4，隊列中等待執行的任務數目：0，已執行玩别的任務數目：0

正在執行task 3

線程池中線程數目：5，隊列中等待執行的任務數目：0，已執行玩别的任務數目：0

正在執行task 4

線程池中線程數目：5，隊列中等待執行的任務數目：1，已執行玩别的任務數目：0

線程池中線程數目：5，隊列中等待執行的任務數目：2，已執行玩别的任務數目：0

線程池中線程數目：5，隊列中等待執行的任務數目：3，已執行玩别的任務數目：0

線程池中線程數目：5，隊列中等待執行的任務數目：4，已執行玩别的任務數目：0

線程池中線程數目：5，隊列中等待執行的任務數目：5，已執行玩别的任務數目：0

線程池中線程數目：6，隊列中等待執行的任務數目：5，已執行玩别的任務數目：0

正在執行task 10

線程池中線程數目：7，隊列中等待執行的任務數目：5，已執行玩别的任務數目：0

正在執行task 11

線程池中線程數目：8，隊列中等待執行的任務數目：5，已執行玩别的任務數目：0

正在執行task 12

線程池中線程數目：9，隊列中等待執行的任務數目：5，已執行玩别的任務數目：0

正在執行task 13

線程池中線程數目：10，隊列中等待執行的任務數目：5，已執行玩别的任務數目：0

正在執行task 14

task 3執行完畢

task 0執行完畢

task 2執行完畢

task 1執行完畢

正在執行task 8

正在執行task 7

正在執行task 6

正在執行task 5

task 4執行完畢

task 10執行完畢

task 11執行完畢

task 13執行完畢

task 12執行完畢

正在執行task 9

task 14執行完畢

task 8執行完畢

task 5執行完畢

task 7執行完畢

task 6執行完畢

task 9執行完畢

從執行結果可以看出，當線程池中線程的數目大于5時，便将任務放入任務緩存隊列裡面，當任務緩存隊列滿了之後，便建立新的線程。如果上面程式中，将for循環中改成執行20個任務，就會抛出任務拒絕異常了。

不過在java doc中，并不提倡我們直接使用threadpoolexecutor，而是使用executors類中提供的幾個靜态方法來建立線程池：

executors.newcachedthreadpool(); //建立一個緩沖池，緩沖池容量大小為integer.max_value

executors.newsinglethreadexecutor(); //建立容量為1的緩沖池

executors.newfixedthreadpool(int); //建立固定容量大小的緩沖池

下面是這三個靜态方法的具體實作;

public static executorservice newfixedthreadpool(int nthreads) {

public static executorservice newsinglethreadexecutor() {

public static executorservice newcachedthreadpool() {

　　從它們的具體實作來看，它們實際上也是調用了threadpoolexecutor，隻不過參數都已配置好了。

　　newfixedthreadpool建立的線程池corepoolsize和maximumpoolsize值是相等的，它使用的linkedblockingqueue；

　　newsinglethreadexecutor将corepoolsize和maximumpoolsize都設定為1，也使用的linkedblockingqueue；

　　newcachedthreadpool将corepoolsize設定為0，将maximumpoolsize設定為integer.max_value，使用的synchronousqueue，也就是說來了任務就建立線程運作，當線程空閑超過60秒，就銷毀線程。

　　實際中，如果executors提供的三個靜态方法能滿足要求，就盡量使用它提供的三個方法，因為自己去手動配置threadpoolexecutor的參數有點麻煩，要根據實際任務的類型和數量來進行配置。

　　另外，如果threadpoolexecutor達不到要求，可以自己繼承threadpoolexecutor類進行重寫。

四.如何合理配置線程池的大小

　　本節來讨論一個比較重要的話題：如何合理配置線程池大小，僅供參考。

　　一般需要根據任務的類型來配置線程池大小：

　　如果是cpu密集型任務，就需要盡量壓榨cpu，參考值可以設為 ncpu+1

　　如果是io密集型任務，參考值可以設定為2*ncpu

　　當然，這隻是一個參考值，具體的設定還需要根據實際情況進行調整，比如可以先将線程池大小設定為參考值，再觀察任務運作情況和系統負載、資源使用率來進行适當調整。