前言

多線程的異步執行方式，雖然能夠最大限度發揮多核計算機的計算能力，但是如果不加控制，反而會對系統造成負擔。線程本身也要占用內存空間，大量的線程會占用內存資源并且可能會導致Out of Memory。即便沒有這樣的情況，大量的線程回收也會給GC帶來很大的壓力。

為了避免重復的創建線程，線程池的出現可以讓線程進行復用。通俗點講，當有工作來，就會向線程池拿一個線程，當工作完成后，并不是直接關閉線程，而是將這個線程歸還給線程池供其他任務使用。

接下來從總體到細致的方式，來共同探討線程池。

總體的架構

來看Executor的框架圖：

?

接口：Executor,CompletionService,ExecutorService，ScheduledExecutorService

抽象類：AbstractExecutorService

實現類：ExecutorCompletionService，ThreadPoolExecutor，ScheduledThreadPoolExecutor

從圖中就可以看到主要的方法，本文主要討論的是ThreadPoolExecutor

研讀ThreadPoolExecutor

看一下該類的構造器：

public ThreadPoolExecutor(int paramInt1, int paramInt2, long paramLong, TimeUnit paramTimeUnit,BlockingQueue<Runnable> paramBlockingQueue, ThreadFactory paramThreadFactory,RejectedExecutionHandler paramRejectedExecutionHandler) {this.ctl = new AtomicInteger(ctlOf(-536870912, 0));this.mainLock = new ReentrantLock();this.workers = new HashSet();this.termination = this.mainLock.newCondition();if ((paramInt1 < 0) || (paramInt2 <= 0) || (paramInt2 < paramInt1) || (paramLong < 0L))throw new IllegalArgumentException();if ((paramBlockingQueue == null) || (paramThreadFactory == null) || (paramRejectedExecutionHandler == null))throw new NullPointerException();this.corePoolSize = paramInt1;this.maximumPoolSize = paramInt2;this.workQueue = paramBlockingQueue;this.keepAliveTime = paramTimeUnit.toNanos(paramLong);this.threadFactory = paramThreadFactory;this.handler = paramRejectedExecutionHandler;}

?

corePoolSize?:線程池的核心池大小，在創建線程池之后，線程池默認沒有任何線程。

當有任務過來的時候才會去創建創建線程執行任務。換個說法，線程池創建之后，線程池中的線程數為0，當任務過來就會創建一個線程去執行，直到線程數達到corePoolSize?之后，就會被到達的任務放在隊列中。（注意是到達的任務）。換句更精煉的話：corePoolSize?表示允許線程池中允許同時運行的最大線程數。

如果執行了線程池的prestartAllCoreThreads()方法，線程池會提前創建并啟動所有核心線程。

maximumPoolSize?:線程池允許的最大線程數，他表示最大能創建多少個線程。maximumPoolSize肯定是大于等于corePoolSize。

keepAliveTime?:表示線程沒有任務時最多保持多久然后停止。默認情況下，只有線程池中線程數大于corePoolSize?時，keepAliveTime?才會起作用。換句話說，當線程池中的線程數大于corePoolSize，并且一個線程空閑時間達到了keepAliveTime，那么就是shutdown。

Unit:keepAliveTime?的單位。

workQueue?：一個阻塞隊列，用來存儲等待執行的任務，當線程池中的線程數超過它的corePoolSize的時候，線程會進入阻塞隊列進行阻塞等待。通過workQueue，線程池實現了阻塞功能

threadFactory?：線程工廠，用來創建線程。

handler?:表示當拒絕處理任務時的策略。

任務緩存隊列

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

workQueue的類型為BlockingQueue<Runnable>，通常可以取下面三種類型：

1）有界任務隊列ArrayBlockingQueue：基于數組的先進先出隊列，此隊列創建時必須指定大小；

2）無界任務隊列LinkedBlockingQueue：基于鏈表的先進先出隊列，如果創建時沒有指定此隊列大小，則默認為Integer.MAX_VALUE；

3）直接提交隊列synchronousQueue：這個隊列比較特殊，它不會保存提交的任務，而是將直接新建一個線程來執行新來的任務。

拒絕策略

AbortPolicy:丟棄任務并拋出RejectedExecutionException

CallerRunsPolicy：只要線程池未關閉，該策略直接在調用者線程中，運行當前被丟棄的任務。顯然這樣做不會真的丟棄任務，但是，任務提交線程的性能極有可能會急劇下降。

DiscardOldestPolicy：丟棄隊列中最老的一個請求，也就是即將被執行的一個任務，并嘗試再次提交當前任務。

DiscardPolicy：丟棄任務，不做任何處理。

線程池的任務處理策略：

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

如果當前線程池中的線程數目>=corePoolSize，則每來一個任務，會嘗試將其添加到任務緩存隊列當中，若添加成功，則該任務會等待空閑線程將其取出去執行；若添加失敗（一般來說是任務緩存隊列已滿），則會嘗試創建新的線程去執行這個任務；如果當前線程池中的線程數目達到maximumPoolSize，則會采取任務拒絕策略進行處理；

如果線程池中的線程數量大于 corePoolSize時，如果某線程空閑時間超過keepAliveTime，線程將被終止，直至線程池中的線程數目不大于corePoolSize；如果允許為核心池中的線程設置存活時間，那么核心池中的線程空閑時間超過keepAliveTime，線程也會被終止。

線程池的關閉

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

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

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

源碼分析

首先來看最核心的execute方法，這個方法在AbstractExecutorService中并沒有實現，從Executor接口，直到ThreadPoolExecutor才實現了改方法，

ExecutorService中的submit(),invokeAll(),invokeAny()都是調用的execute方法，所以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.* 如果正在運行的線程數小于corePoolSize，那么將調用addWorker 方法來創建一個新的線程，并將該任務作為新線程的第一個任務來執行。
　　　　　　 當然，在創建線程之前會做原子性質的檢查，如果條件不允許，則不創建線程來執行任務，并返回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.
　　　　　　 如果任務無法入隊列（隊列滿了），那么我們將嘗試新開啟一個線程（從corepoolsize到擴充到maximum），如果失敗了，那么可以確定原因，要么是
　　　　　　 線程池關閉了或者飽和了（達到maximum），所以我們執行拒絕策略。
*/
　　　　
　　　　// 1.當前線程數量小于corePoolSize，則創建并啟動線程。int c = ctl.get();if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))
　　　　　　　　// 成功，則返回 return;c = ctl.get();}
　　　　// 2.步驟1失敗，則嘗試進入阻塞隊列，if (isRunning(c) && workQueue.offer(command)) {
　　　　　　　// 入隊列成功，檢查線程池狀態，如果狀態部署RUNNING而且remove成功，則拒絕任務int recheck = ctl.get();if (! isRunning(recheck) && remove(command))reject(command);
　　　　　　　// 如果當前worker數量為0，通過addWorker(null, false)創建一個線程，其任務為nullelse if (workerCountOf(recheck) == 0)addWorker(null, false);}
　　　　// 3. 步驟1和2失敗，則嘗試將線程池的數量有corePoolSize擴充至maxPoolSize，如果失敗，則拒絕任務else if (!addWorker(command, false))reject(command);}

?

?相信看了代碼也是一臉懵,接下來用一個流程圖來講一講，他究竟干了什么事：

結合上面的流程圖來逐行解析，首先前面進行空指針檢查，

wonrkerCountOf()方法能夠取得當前線程池中的線程的總數，取得當前線程數與核心池大小比較，

• 如果小于，將通過addWorker()方法調度執行。
• 如果大于核心池大小，那么就提交到等待隊列。
• 如果進入等待隊列失敗，則會將任務直接提交給線程池。
• 如果線程數達到最大線程數，那么就提交失敗，執行拒絕策略。

?

excute()方法中添加任務的方式是使用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;// 內層的循環，任務是將worker數量加1for (;;) {int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get(); // Re-read ctlif (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}// worker加1后，接下來將woker添加到HashSet<Worker>中，并啟動workerboolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {final ReentrantLock mainLock = this.mainLock;w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int c = ctl.get();int rs = runStateOf(c);if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}
　　　　　　　　　// 如果往HashSet<Worker>添加成功，則啟動該線程if (workerAdded) {t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;}

addWorker(Runnable firstTask,?boolean?core)的主要任務是創建并啟動線程。

他會根據當前線程的狀態和給定的值（core or maximum）來判斷是否可以創建一個線程。

addWorker共有四種傳參方式。execute使用了其中三種，分別為:

1.addWorker(paramRunnable,?true)

線程數小于corePoolSize時，放一個需要處理的task進Workers Set。如果Workers Set長度超過corePoolSize，就返回false.

2.addWorker(null,?false)

放入一個空的task進workers Set，長度限制是maximumPoolSize。這樣一個task為空的worker在線程執行的時候會去任務隊列里拿任務，這樣就相當于創建了一個新的線程，只是沒有馬上分配任務。

3.addWorker(paramRunnable,?false)

當隊列被放滿時，就嘗試將這個新來的task直接放入Workers Set，而此時Workers Set的長度限制是maximumPoolSize。如果線程池也滿了的話就返回false.

?

還有一種情況是execute()方法沒有使用的

addWorker(null,?true)

這個方法就是放一個null的task進Workers Set，而且是在小于corePoolSize時，如果此時Set中的數量已經達到corePoolSize那就返回false，什么也不干。實際使用中是在prestartAllCoreThreads()方法，這個方法用來為線程池預先啟動corePoolSize個worker等待從workQueue中獲取任務執行。

執行流程：

1、判斷線程池當前是否為可以添加worker線程的狀態，可以則繼續下一步，不可以return false：
??? A、線程池狀態>shutdown，可能為stop、tidying、terminated，不能添加worker線程
??? B、線程池狀態==shutdown，firstTask不為空，不能添加worker線程，因為shutdown狀態的線程池不接收新任務
??? C、線程池狀態==shutdown，firstTask==null，workQueue為空，不能添加worker線程，因為firstTask為空是為了添加一個沒有任務的線程再從workQueue獲取task，而workQueue為? 　　　　空，說明添加無任務線程已經沒有意義
2、線程池當前線程數量是否超過上限（corePoolSize 或 maximumPoolSize），超過了return false，沒超過則對workerCount+1，繼續下一步
3、在線程池的ReentrantLock保證下，向Workers Set中添加新創建的worker實例，添加完成后解鎖，并啟動worker線程，如果這一切都成功了，return true，如果添加worker入Set失敗或啟動失敗，調用addWorkerFailed()邏輯

?

常見的四種線程池

newFixedThreadPool

public static ExecutorService newFixedThreadPool(int var0) {return new ThreadPoolExecutor(var0, var0, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());} public static ExecutorService newFixedThreadPool(int var0, ThreadFactory var1) {
return new ThreadPoolExecutor(var0, var0, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), var1);
}

固定大小的線程池，可以指定線程池的大小，該線程池corePoolSize和maximumPoolSize相等，阻塞隊列使用的是LinkedBlockingQueue，大小為整數最大值。

該線程池中的線程數量始終不變，當有新任務提交時，線程池中有空閑線程則會立即執行，如果沒有，則會暫存到阻塞隊列。對于固定大小的線程池，不存在線程數量的變化。同時使用無界的LinkedBlockingQueue來存放執行的任務。當任務提交十分頻繁的時候，LinkedBlockingQueue

迅速增大，存在著耗盡系統資源的問題。而且在線程池空閑時，即線程池中沒有可運行任務時，它也不會釋放工作線程，還會占用一定的系統資源，需要shutdown。

newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {return new Executors.FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()));}public static ExecutorService newSingleThreadExecutor(ThreadFactory var0) {return new Executors.FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), var0));}

單個線程線程池，只有一個線程的線程池，阻塞隊列使用的是LinkedBlockingQueue,若有多余的任務提交到線程池中，則會被暫存到阻塞隊列，待空閑時再去執行。按照先入先出的順序執行任務。

newCachedThreadPool

public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, 2147483647, 60L, TimeUnit.SECONDS, new SynchronousQueue());}public static ExecutorService newCachedThreadPool(ThreadFactory var0) {return new ThreadPoolExecutor(0, 2147483647, 60L, TimeUnit.SECONDS, new SynchronousQueue(), var0);}

緩存線程池，緩存的線程默認存活60秒。線程的核心池corePoolSize大小為0，核心池最大為Integer.MAX_VALUE,阻塞隊列使用的是SynchronousQueue。是一個直接提交的阻塞隊列，??? 他總會迫使線程池增加新的線程去執行新的任務。在沒有任務執行時，當線程的空閑時間超過keepAliveTime（60秒），則工作線程將會終止被回收，當提交新任務時，如果沒有空閑線程，則創建新線程執行任務，會導致一定的系統開銷。如果同時又大量任務被提交，而且任務執行的時間不是特別快，那么線程池便會新增出等量的線程池處理任務，這很可能會很快耗盡系統的資源。

newScheduledThreadPool

public static ScheduledExecutorService newScheduledThreadPool(int var0) {return new ScheduledThreadPoolExecutor(var0);}public static ScheduledExecutorService newScheduledThreadPool(int var0, ThreadFactory var1) {return new ScheduledThreadPoolExecutor(var0, var1);}

定時線程池，該線程池可用于周期性地去執行任務，通常用于周期性的同步數據。

scheduleAtFixedRate:是以固定的頻率去執行任務，周期是指每次執行任務成功執行之間的間隔。

schedultWithFixedDelay:是以固定的延時去執行任務，延時是指上一次執行成功之后和下一次開始執行的之前的時間。

?

使用實例 ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ??

newFixedThreadPool實例：

?View Code

?

newCachedThreadPool實例：

?View Code

這里沒用調用shutDown方法，這里可以發現過60秒之后，會自動釋放資源。

?

newSingleThreadExecutor

?View Code

這里需要注意一點，newSingleThreadExecutor和newFixedThreadPool一樣，在線程池中沒有任務時可執行，也不會釋放系統資源的，所以需要shudown。

?

newScheduledThreadPool

?View Code

?

最后雜談 ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?

如何選擇線程池數量

線程池的大小決定著系統的性能，過大或者過小的線程池數量都無法發揮最優的系統性能。

當然線程池的大小也不需要做的太過于精確，只需要避免過大和過小的情況。一般來說，確定線程池的大小需要考慮CPU的數量，內存大小，任務是計算密集型還是IO密集型等因素

NCPU = CPU的數量

UCPU = 期望對CPU的使用率 0 ≤ UCPU ≤ 1

W/C = 等待時間與計算時間的比率

如果希望處理器達到理想的使用率，那么線程池的最優大小為：

線程池大小=NCPU *UCPU(1+W/C)

在Java中使用

int ncpus = Runtime.getRuntime().availableProcessors();

獲取CPU的數量。

?

線程池工廠

Executors的線程池如果不指定線程工廠會使用Executors中的DefaultThreadFactory,默認線程池工廠創建的線程都是非守護線程。

使用自定義的線程工廠可以做很多事情，比如可以跟蹤線程池在何時創建了多少線程，也可以自定義線程名稱和優先級。如果將

新建的線程都設置成守護線程，當主線程退出后，將會強制銷毀線程池。

下面這個例子，記錄了線程的創建，并將所有的線程設置成守護線程。

?

?View Code

?

擴展線程池

ThreadPoolExecutor是可以拓展的，它提供了幾個可以在子類中改寫的方法：beforeExecute,afterExecute和terimated。

在執行任務的線程中將調用beforeExecute和afterExecute,這些方法中還可以添加日志，計時，監視或統計收集的功能，

還可以用來輸出有用的調試信息，幫助系統診斷故障。下面是一個擴展線程池的例子：

?

?View Code

?

線程池的正確使用

以下阿里編碼規范里面說的一段話：

線程池不允許使用Executors去創建，而是通過ThreadPoolExecutor的方式，這樣的處理方式讓寫的同學更加明確線程池的運行規則，規避資源耗盡的風險。 說明：Executors各個方法的弊端：
1）newFixedThreadPool和newSingleThreadExecutor:
??主要問題是堆積的請求處理隊列可能會耗費非常大的內存，甚至OOM。
2）newCachedThreadPool和newScheduledThreadPool:
??主要問題是線程數最大數是Integer.MAX_VALUE，可能會創建數量非常多的線程，甚至OOM。

?

手動創建線程池有幾個注意點

1.任務獨立。如何任務依賴于其他任務，那么可能產生死鎖。例如某個任務等待另一個任務的返回值或執行結果，那么除非線程池足夠大，否則將發生線程饑餓死鎖。

2.合理配置阻塞時間過長的任務。如果任務阻塞時間過長，那么即使不出現死鎖，線程池的性能也會變得很糟糕。在Java并發包里可阻塞方法都同時定義了限時方式和不限時方式。例如

Thread.join,BlockingQueue.put,CountDownLatch.await等，如果任務超時，則標識任務失敗，然后中止任務或者將任務放回隊列以便隨后執行，這樣，無論任務的最終結果是否成功，這種辦法都能夠保證任務總能繼續執行下去。

3.設置合理的線程池大小。只需要避免過大或者過小的情況即可，上文的公式線程池大小=NCPU *UCPU(1+W/C)。

4.選擇合適的阻塞隊列。newFixedThreadPool和newSingleThreadExecutor都使用了無界的阻塞隊列，無界阻塞隊列會有消耗很大的內存，如果使用了有界阻塞隊列，它會規避內存占用過大的問題，但是當任務填滿有界阻塞隊列，新的任務該怎么辦？在使用有界隊列是，需要選擇合適的拒絕策略，隊列的大小和線程池的大小必須一起調節。對于非常大的或者無界的線程池，可以使用SynchronousQueue來避免任務排隊，以直接將任務從生產者提交到工作者線程。

?

下面是Thrift框架處理socket任務所使用的一個線程池，可以看一下FaceBook的工程師是如何自定義線程池的。

private static ExecutorService createDefaultExecutorService(Args args) {SynchronousQueue executorQueue = new SynchronousQueue();return new ThreadPoolExecutor(args.minWorkerThreads, args.maxWorkerThreads, 60L, TimeUnit.SECONDS,executorQueue);}

總結：

查看以備份

參考資料

《實戰Java》高并發程序設計

《Java Concurrency in Practice》

Java線程池ThreadPoolExecutor使用和分析(二)

轉載于:https://www.cnblogs.com/wwmiert/p/9441289.html

總結

以上是生活随笔為你收集整理的Java线程池理解及用法的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。