Java執行緒池使用說明【比較全面】
一簡介
執行緒的使用在Java中佔有極其重要的地位，在jdk1.4極其之前的jdk版本中，關於執行緒池的使用是極其簡陋的。在jdk1.5之後這一情況有了很大的改觀。Jdk1.5之後加入了java.util.concurrent包，這個包中主要介紹java中執行緒以及執行緒池的使用。為我們在開發中處理執行緒的問題提供了非常大的幫助。二：執行緒池
執行緒池的作用：
執行緒池作用就是限制系統中執行執行緒的數量。
根據系統的環境情況，可以自動或手動設定執行緒數量，達到執行的最佳效果；少了浪費了系統資源，多了造成系統擁擠效率不高。用執行緒池控制執行緒數量，其他執行緒排隊等候。一個任務執行完畢，再從佇列的中取最前面的任務開始執行。若佇列中沒有等待程序，執行緒池的這一資源處於等待。當一個新任務需要執行時，如果執行緒池中有等待的工作執行緒，就可以開始運行了；否則進入等待佇列。
為什麼要用執行緒池:
1.減少了建立和銷燬執行緒的次數，每個工作執行緒都可以被重複利用，可執行多個任務。
2.可以根據系統的承受能力，調整執行緒池中工作線執行緒的數目，防止因為消耗過多的記憶體，而把伺服器累趴下(每個執行緒需要大約1MB記憶體，執行緒開的越多，消耗的記憶體也就越大，最後宕機)。
Java裡面執行緒池的頂級介面是Executor，但是嚴格意義上講Executor並不是一個執行緒池，而只是一個執行執行緒的工具。真正的執行緒池介面是ExecutorService。
比較重要的幾個類：
|
ExecutorService |
真正的執行緒池介面。 |
|
ScheduledExecutorService |
能和Timer/TimerTask類似，解決那些需要任務重複執行的問題。 |
|
ThreadPoolExecutor |
ExecutorService的預設實現。 |
|
ScheduledThreadPoolExecutor |
繼承ThreadPoolExecutor的ScheduledExecutorService介面實現，週期性任務排程的類實現。 |
要配置一個執行緒池是比較複雜的，尤其是對於執行緒池的原理不是很清楚的情況下，很有可能配置的執行緒池不是較優的，因此在Executors類裡面提供了一些靜態工廠，生成一些常用的執行緒池。
1. newSingleThreadExecutor
建立一個單執行緒的執行緒池。這個執行緒池只有一個執行緒在工作，也就是相當於單執行緒序列執行所有任務。如果這個唯一的執行緒因為異常結束，那麼會有一個新的執行緒來替代它。此執行緒池保證所有任務的執行順序按照任務的提交順序執行。
2.newFixedThreadPool
建立固定大小的執行緒池。每次提交一個任務就建立一個執行緒，直到執行緒達到執行緒池的最大大小。執行緒池的大小一旦達到最大值就會保持不變，如果某個執行緒因為執行異常而結束，那麼執行緒池會補充一個新執行緒。
3. newCachedThreadPool
建立一個可快取的執行緒池。如果執行緒池的大小超過了處理任務所需要的執行緒，
那麼就會回收部分空閒（60秒不執行任務）的執行緒，當任務數增加時，此執行緒池又可以智慧的新增新執行緒來處理任務。此執行緒池不會對執行緒池大小做限制，執行緒池大小完全依賴於作業系統（或者說JVM）能夠建立的最大執行緒大小。
4.newScheduledThreadPool
建立一個大小無限的執行緒池。此執行緒池支援定時以及週期性執行任務的需求。
例項
1：newSingleThreadExecutor
MyThread.java
|
publicclassMyThreadextends Thread { @Override publicvoid run() { System.out.println(Thread.currentThread().getName() + "正在執行。。。"); } } |
TestSingleThreadExecutor.java
|
publicclassTestSingleThreadExecutor { publicstaticvoid main(String[] args) { //建立一個可重用固定執行緒數的執行緒池 ExecutorService pool = Executors. newSingleThreadExecutor(); //建立實現了Runnable介面物件，Thread物件當然也實現了Runnable介面 Thread t1 =new MyThread(); Thread t2 =new MyThread(); Thread t3 =new MyThread(); Thread t4 =new MyThread(); Thread t5 =new MyThread(); //將執行緒放入池中進行執行 pool.execute(t1); pool.execute(t2); pool.execute(t3); pool.execute(t4); pool.execute(t5); //關閉執行緒池 pool.shutdown(); } } |
輸出結果
|
pool-1-thread-1正在執行。。。 pool-1-thread-1正在執行。。。 pool-1-thread-1正在執行。。。 pool-1-thread-1正在執行。。。 pool-1-thread-1正在執行。。。 |
2newFixedThreadPool
TestFixedThreadPool.Java
|
publicclass TestFixedThreadPool { publicstaticvoid main(String[] args) { //建立一個可重用固定執行緒數的執行緒池 ExecutorService pool = Executors.newFixedThreadPool(2); //建立實現了Runnable介面物件，Thread物件當然也實現了Runnable介面 Thread t1 =new MyThread(); Thread t2 =new MyThread(); Thread t3 =new MyThread(); Thread t4 =new MyThread(); Thread t5 =new MyThread(); //將執行緒放入池中進行執行 pool.execute(t1); pool.execute(t2); pool.execute(t3); pool.execute(t4); pool.execute(t5); //關閉執行緒池 pool.shutdown(); } } |
輸出結果
|
pool-1-thread-1正在執行。。。 pool-1-thread-2正在執行。。。 pool-1-thread-1正在執行。。。 pool-1-thread-2正在執行。。。 pool-1-thread-1正在執行。。。 |
3 newCachedThreadPool
TestCachedThreadPool.java
|
publicclass TestCachedThreadPool { publicstaticvoid main(String[] args) { //建立一個可重用固定執行緒數的執行緒池 ExecutorService pool = Executors.newCachedThreadPool(); //建立實現了Runnable介面物件，Thread物件當然也實現了Runnable介面 Thread t1 =new MyThread(); Thread t2 =new MyThread(); Thread t3 =new MyThread(); Thread t4 =new MyThread(); Thread t5 =new MyThread(); //將執行緒放入池中進行執行 pool.execute(t1); pool.execute(t2); pool.execute(t3); pool.execute(t4); pool.execute(t5); //關閉執行緒池 pool.shutdown(); } } |
輸出結果：
|
pool-1-thread-2正在執行。。。 pool-1-thread-4正在執行。。。 pool-1-thread-3正在執行。。。 pool-1-thread-1正在執行。。。 pool-1-thread-5正在執行。。。 |
4newScheduledThreadPool
TestScheduledThreadPoolExecutor.java
|
publicclass TestScheduledThreadPoolExecutor { publicstaticvoid main(String[] args) { ScheduledThreadPoolExecutor exec =new ScheduledThreadPoolExecutor(1); exec.scheduleAtFixedRate(new Runnable() {//每隔一段時間就觸發異常 @Override publicvoid run() { //throw new RuntimeException(); System.out.println("================"); } }, 1000, 5000, TimeUnit.MILLISECONDS); exec.scheduleAtFixedRate(new Runnable() {//每隔一段時間列印系統時間，證明兩者是互不影響的 @Override publicvoid run() { System.out.println(System.nanoTime()); } }, 1000, 2000, TimeUnit.MILLISECONDS); } } |
輸出結果
|
================ 8384644549516 8386643829034 8388643830710 ================ 8390643851383 8392643879319 8400643939383 |
三：ThreadPoolExecutor詳解
ThreadPoolExecutor的完整構造方法的簽名是：ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) .
corePoolSize - 池中所儲存的執行緒數，包括空閒執行緒。
maximumPoolSize-池中允許的最大執行緒數。
keepAliveTime - 當執行緒數大於核心時，此為終止前多餘的空閒執行緒等待新任務的最長時間。
unit - keepAliveTime 引數的時間單位。
workQueue - 執行前用於保持任務的佇列。此佇列僅保持由 execute方法提交的 Runnable任務。
threadFactory - 執行程式建立新執行緒時使用的工廠。
handler - 由於超出執行緒範圍和佇列容量而使執行被阻塞時所使用的處理程式。
ThreadPoolExecutor是Executors類的底層實現。
在JDK幫助文件中，有如此一段話：
“強烈建議程式設計師使用較為方便的Executors工廠方法Executors.newCachedThreadPool()（無界執行緒池，可以進行自動執行緒回收）、Executors.newFixedThreadPool(int)（固定大小執行緒池）Executors.newSingleThreadExecutor()（單個後臺執行緒）
它們均為大多數使用場景預定義了設定。”
下面介紹一下幾個類的原始碼：
ExecutorService newFixedThreadPool (int nThreads):固定大小執行緒池。
可以看到，corePoolSize和maximumPoolSize的大小是一樣的（實際上，後面會介紹，如果使用無界queue的話maximumPoolSize引數是沒有意義的），keepAliveTime和unit的設值表名什麼？-就是該實現不想keep alive！最後的BlockingQueue選擇了LinkedBlockingQueue，該queue有一個特點，他是無界的。
|
1. public static ExecutorService newFixedThreadPool(int nThreads) { 2. return new ThreadPoolExecutor(nThreads, nThreads, 3. 0L, TimeUnit.MILLISECONDS, 4. new LinkedBlockingQueue<Runnable>()); 5. } |
ExecutorService newSingleThreadExecutor()：單執行緒
|
1. public static ExecutorService newSingleThreadExecutor() { 2. return new FinalizableDelegatedExecutorService 3. (new ThreadPoolExecutor(1, 1, 4. 0L, TimeUnit.MILLISECONDS, 5. new LinkedBlockingQueue<Runnable>())); 6. } |
ExecutorService newCachedThreadPool()：無界執行緒池，可以進行自動執行緒回收
這個實現就有意思了。首先是無界的執行緒池，所以我們可以發現maximumPoolSize為big big。其次BlockingQueue的選擇上使用SynchronousQueue。可能對於該BlockingQueue有些陌生，簡單說：該QUEUE中，每個插入操作必須等待另一個執行緒的對應移除操作。
|
1. public static ExecutorService newCachedThreadPool() { 2. return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 3. 60L, TimeUnit.SECONDS, 4. new SynchronousQueue<Runnable>());
|
先從BlockingQueue<Runnable> workQueue這個入參開始說起。在JDK中，其實已經說得很清楚了，一共有三種類型的queue。
所有BlockingQueue都可用於傳輸和保持提交的任務。可以使用此佇列與池大小進行互動：
如果執行的執行緒少於 corePoolSize，則 Executor始終首選新增新的執行緒，而不進行排隊。（如果當前執行的執行緒小於corePoolSize，則任務根本不會存放，新增到queue中，而是直接抄傢伙（thread）開始執行）
如果執行的執行緒等於或多於 corePoolSize，則 Executor始終首選將請求加入佇列，而不新增新的執行緒。
如果無法將請求加入佇列，則建立新的執行緒，除非建立此執行緒超出 maximumPoolSize，在這種情況下，任務將被拒絕。
queue上的三種類型。
排隊有三種通用策略：
直接提交。工作佇列的預設選項是 SynchronousQueue，它將任務直接提交給執行緒而不保持它們。在此，如果不存在可用於立即執行任務的執行緒，則試圖把任務加入佇列將失敗，因此會構造一個新的執行緒。此策略可以避免在處理可能具有內部依賴性的請求集時出現鎖。直接提交通常要求無界 maximumPoolSizes 以避免拒絕新提交的任務。當命令以超過佇列所能處理的平均數連續到達時，此策略允許無界執行緒具有增長的可能性。
無界佇列。使用無界佇列（例如，不具有預定義容量的 LinkedBlockingQueue）將導致在所有 corePoolSize 執行緒都忙時新任務在佇列中等待。這樣，建立的執行緒就不會超過 corePoolSize。（因此，maximumPoolSize的值也就無效了。）當每個任務完全獨立於其他任務，即任務執行互不影響時，適合於使用無界佇列；例如，在 Web頁伺服器中。這種排隊可用於處理瞬態突發請求，當命令以超過佇列所能處理的平均數連續到達時，此策略允許無界執行緒具有增長的可能性。
有界佇列。當使用有限的 maximumPoolSizes時，有界佇列（如 ArrayBlockingQueue）有助於防止資源耗盡，但是可能較難調整和控制。佇列大小和最大池大小可能需要相互折衷：使用大型佇列和小型池可以最大限度地降低 CPU使用率、作業系統資源和上下文切換開銷，但是可能導致人工降低吞吐量。如果任務頻繁阻塞（例如，如果它們是 I/O邊界），則系統可能為超過您許可的更多執行緒安排時間。使用小型佇列通常要求較大的池大小，CPU使用率較高，但是可能遇到不可接受的排程開銷，這樣也會降低吞吐量。
BlockingQueue的選擇。
例子一：使用直接提交策略，也即SynchronousQueue。
首先SynchronousQueue是無界的，也就是說他存數任務的能力是沒有限制的，但是由於該Queue本身的特性，在某次新增元素後必須等待其他執行緒取走後才能繼續新增。在這裡不是核心執行緒便是新建立的執行緒，但是我們試想一樣下，下面的場景。
我們使用一下引數構造ThreadPoolExecutor：
1. new ThreadPoolExecutor(
2. 2, 3, 30, TimeUnit.SECONDS,
3. new SynchronousQueue<Runnable>(),
4. new RecorderThreadFactory("CookieRecorderPool"),
- new ThreadPoolExecutor.CallerRunsPolicy());
new ThreadPoolExecutor(
2, 3, 30, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
new RecorderThreadFactory("CookieRecorderPool"),
new ThreadPoolExecutor.CallerRunsPolicy());
當核心執行緒已經有2個正在執行.
- 此時繼續來了一個任務（A），根據前面介紹的“如果執行的執行緒等於或多於 corePoolSize，則 Executor始終首選將請求加入佇列，而不新增新的執行緒。”,所以A被新增到queue中。
- 又來了一個任務（B），且核心2個執行緒還沒有忙完，OK，接下來首先嚐試1中描述，但是由於使用的SynchronousQueue，所以一定無法加入進去。
- 此時便滿足了上面提到的“如果無法將請求加入佇列，則建立新的執行緒，除非建立此執行緒超出maximumPoolSize，在這種情況下，任務將被拒絕。”，所以必然會新建一個執行緒來執行這個任務。
- 暫時還可以，但是如果這三個任務都還沒完成，連續來了兩個任務，第一個新增入queue中，後一個呢？queue中無法插入，而執行緒數達到了maximumPoolSize，所以只好執行異常策略了。
所以在使用SynchronousQueue通常要求maximumPoolSize是無界的，這樣就可以避免上述情況發生（如果希望限制就直接使用有界佇列）。對於使用SynchronousQueue的作用jdk中寫的很清楚：此策略可以避免在處理可能具有內部依賴性的請求集時出現鎖。
什麼意思？如果你的任務A1，A2有內部關聯，A1需要先執行，那麼先提交A1，再提交A2，當使用SynchronousQueue我們可以保證，A1必定先被執行，在A1麼有被執行前，A2不可能新增入queue中。
例子二：使用無界佇列策略，即LinkedBlockingQueue
這個就拿newFixedThreadPool來說，根據前文提到的規則：
如果執行的執行緒少於 corePoolSize，則 Executor 始終首選新增新的執行緒，而不進行排隊。那麼當任務繼續增加，會發生什麼呢？
如果執行的執行緒等於或多於 corePoolSize，則 Executor 始終首選將請求加入佇列，而不新增新的執行緒。OK，此時任務變加入佇列之中了，那什麼時候才會新增新執行緒呢？
如果無法將請求加入佇列，則建立新的執行緒，除非建立此執行緒超出 maximumPoolSize，在這種情況下，任務將被拒絕。這裡就很有意思了，可能會出現無法加入佇列嗎？不像SynchronousQueue那樣有其自身的特點，對於無界佇列來說，總是可以加入的（資源耗盡，當然另當別論）。換句說，永遠也不會觸發產生新的執行緒！corePoolSize大小的執行緒數會一直執行，忙完當前的，就從佇列中拿任務開始執行。所以要防止任務瘋長，比如任務執行的實行比較長，而新增任務的速度遠遠超過處理任務的時間，而且還不斷增加，不一會兒就爆了。
例子三：有界佇列，使用ArrayBlockingQueue。
這個是最為複雜的使用，所以JDK不推薦使用也有些道理。與上面的相比，最大的特點便是可以防止資源耗盡的情況發生。
舉例來說，請看如下構造方法：
1. new ThreadPoolExecutor(
2. 2, 4, 30, TimeUnit.SECONDS,
3. new ArrayBlockingQueue<Runnable>(2),
4. new RecorderThreadFactory("CookieRecorderPool"),
5. new ThreadPoolExecutor.CallerRunsPolicy());
new ThreadPoolExecutor(
2, 4, 30, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(2),
new RecorderThreadFactory("CookieRecorderPool"),
new ThreadPoolExecutor.CallerRunsPolicy());
假設，所有的任務都永遠無法執行完。
對於首先來的A,B來說直接執行，接下來，如果來了C,D，他們會被放到queue中，如果接下來再來E,F，則增加執行緒執行E，F。但是如果再來任務，佇列無法再接受了，執行緒數也到達最大的限制了，所以就會使用拒絕策略來處理。
keepAliveTime
jdk中的解釋是：當執行緒數大於核心時，此為終止前多餘的空閒執行緒等待新任務的最長時間。
有點拗口，其實這個不難理解，在使用了“池”的應用中，大多都有類似的引數需要配置。比如資料庫連線池，DBCP中的maxIdle，minIdle引數。
什麼意思？接著上面的解釋，後來向老闆派來的工人始終是“借來的”，俗話說“有借就有還”，但這裡的問題就是什麼時候還了，如果借來的工人剛完成一個任務就還回去，後來發現任務還有，那豈不是又要去借？這一來一往，老闆肯定頭也大死了。
合理的策略：既然借了，那就多借一會兒。直到“某一段”時間後，發現再也用不到這些工人時，便可以還回去了。這裡的某一段時間便是keepAliveTime的含義，TimeUnit為keepAliveTime值的度量。
RejectedExecutionHandler
另一種情況便是，即使向老闆借了工人，但是任務還是繼續過來，還是忙不過來，這時整個隊伍只好拒絕接受了。
RejectedExecutionHandler介面提供了對於拒絕任務的處理的自定方法的機會。在ThreadPoolExecutor中已經預設包含了4中策略，因為原始碼非常簡單，這裡直接貼出來。
CallerRunsPolicy：執行緒呼叫執行該任務的 execute 本身。此策略提供簡單的反饋控制機制，能夠減緩新任務的提交速度。
1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
2. if (!e.isShutdown()) {
3. r.run();
4. }
5. }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
這個策略顯然不想放棄執行任務。但是由於池中已經沒有任何資源了，那麼就直接使用呼叫該execute的執行緒本身來執行。
AbortPolicy：處理程式遭到拒絕將丟擲執行時RejectedExecutionException
1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
2. throw new RejectedExecutionException();
3. }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException();
}
這種策略直接丟擲異常，丟棄任務。
DiscardPolicy：不能執行的任務將被刪除
1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
2. }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
這種策略和AbortPolicy幾乎一樣，也是丟棄任務，只不過他不丟擲異常。
DiscardOldestPolicy：如果執行程式尚未關閉，則位於工作佇列頭部的任務將被刪除，然後重試執行程式（如果再次失敗，則重複此過程）
1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
2. if (!e.isShutdown()) {
3. e.getQueue().poll();
4. e.execute(r);
5. }
- }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
該策略就稍微複雜一些，在pool沒有關閉的前提下首先丟掉快取在佇列中的最早的任務，然後重新嘗試執行該任務。這個策略需要適當小心。
設想:如果其他執行緒都還在執行，那麼新來任務踢掉舊任務，快取在queue中，再來一個任務又會踢掉queue中最老任務。
總結：
keepAliveTime和maximumPoolSize及BlockingQueue的型別均有關係。如果BlockingQueue是無界的，那麼永遠不會觸發maximumPoolSize，自然keepAliveTime也就沒有了意義。
反之，如果核心數較小，有界BlockingQueue數值又較小，同時keepAliveTime又設的很小，如果任務頻繁，那麼系統就會頻繁的申請回收執行緒。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
轉自：http://blog.csdn.net/sd0902/article/details/8395677