線程的使用在java中占有極其重要的地位，在jdk1.4極其之前的jdk版本中，關于線程池的使用是極其簡陋的。在jdk1.5之后這一情況有了很大的改觀。Jdk1.5之后加入了java.util.concurrent包，這個包中主要介紹java中線程以及線程池的使用。為我們在開發(fā)中處理線程的問題提供了非常大的幫助。
線程池的作用：
線程池作用就是限制系統(tǒng)中執(zhí)行線程的數(shù)量。
根據(jù)系統(tǒng)的環(huán)境情況，可以自動或手動設置線程數(shù)量，達到運行的最佳效果；少了浪費了系統(tǒng)資源，多了造成系統(tǒng)擁擠效率不高。用線程池控制線程數(shù)量，其他線程排隊等候。一個任務執(zhí)行完畢，再從隊列的中取最前面的任務開始執(zhí)行。若隊列中沒有等待進程，線程池的這一資源處于等待。當一個新任務需要運行時，如果線程池中有等待的工作線程，就可以開始運行了；否則進入等待隊列。
為什么要用線程池:
1.減少了創(chuàng)建和銷毀線程的次數(shù)，每個工作線程都可以被重復利用，可執(zhí)行多個任務。
2.可以根據(jù)系統(tǒng)的承受能力，調整線程池中工作線線程的數(shù)目，防止因為消耗過多的內存，而把服務器累趴下(每個線程需要大約1MB內存，線程開的越多，消耗的內存也就越大，最后死機)。
Java里面線程池的頂級接口是Executor，但是嚴格意義上講Executor并不是一個線程池，而只是一個執(zhí)行線程的工具。真正的線程池接口是ExecutorService。
比較重要的幾個類：
| ExecutorService | 真正的線程池接口。 |
| ScheduledExecutorService | 能和Timer/TimerTask類似，解決那些需要任務重復執(zhí)行的問題。 |
| ThreadPoolExecutor | ExecutorService的默認實現(xiàn)。 |
| ScheduledThreadPoolExecutor | 繼承ThreadPoolExecutor的ScheduledExecutorService接口實現(xiàn)，周期性任務調度的類實現(xiàn)。 |
要配置一個線程池是比較復雜的，尤其是對于線程池的原理不是很清楚的情況下，很有可能配置的線程池不是較優(yōu)的，因此在Executors類里面提供了一些靜態(tài)工廠，生成一些常用的線程池。
1. newSingleThreadExecutor
創(chuàng)建一個單線程的線程池。這個線程池只有一個線程在工作，也就是相當于單線程串行執(zhí)行所有任務。如果這個唯一的線程因為異常結束，那么會有一個新的線程來替代它。此線程池保證所有任務的執(zhí)行順序按照任務的提交順序執(zhí)行。
2.newFixedThreadPool
創(chuàng)建固定大小的線程池。每次提交一個任務就創(chuàng)建一個線程，直到線程達到線程池的最大大小。線程池的大小一旦達到最大值就會保持不變，如果某個線程因為執(zhí)行異常而結束，那么線程池會補充一個新線程。
3. newCachedThreadPool
創(chuàng)建一個可緩存的線程池。如果線程池的大小超過了處理任務所需要的線程，
那么就會回收部分空閑（60秒不執(zhí)行任務）的線程，當任務數(shù)增加時，此線程池又可以智能的添加新線程來處理任務。此線程池不會對線程池大小做限制，線程池大小完全依賴于操作系統(tǒng)（或者說JVM）能夠創(chuàng)建的最大線程大小。
4.newScheduledThreadPool
創(chuàng)建一個大小無限的線程池。此線程池支持定時以及周期性執(zhí)行任務的需求。
實例
1：newSingleThreadExecutor
MyThread.java
|
1
2
3
4
5
6
7
8
9
10
11
|
public class MyThread extends Thread { @Override public void run() { System.out.println(Thread.currentThread().getName() + "正在執(zhí)行。。。"); } } |
TestSingleThreadExecutor.java
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
public class TestSingleThreadExecutor { public static void main(String[] args) { //創(chuàng)建一個可重用固定線程數(shù)的線程池 ExecutorService pool = Executors. newSingleThreadExecutor(); //創(chuàng)建實現(xiàn)了Runnable接口對象，Thread對象當然也實現(xiàn)了Runnable接口 Thread t1 = new MyThread(); Thread t2 = new MyThread(); Thread t3 = new MyThread(); Thread t4 = new MyThread(); Thread t5 = new MyThread(); //將線程放入池中進行執(zhí)行 pool.execute(t1); pool.execute(t2); pool.execute(t3); pool.execute(t4); pool.execute(t5); //關閉線程池 pool.shutdown(); } } |
輸出結果：
pool-1-thread-1正在執(zhí)行。。。
pool-1-thread-1正在執(zhí)行。。。
pool-1-thread-1正在執(zhí)行。。。
pool-1-thread-1正在執(zhí)行。。。
pool-1-thread-1正在執(zhí)行。。。
2: newFixedThreadPool
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
public class TestFixedThreadPool { publicstaticvoid main(String[] args) { //創(chuàng)建一個可重用固定線程數(shù)的線程池 ExecutorService pool = Executors.newFixedThreadPool(2); //創(chuàng)建實現(xiàn)了Runnable接口對象，Thread對象當然也實現(xiàn)了Runnable接口 Thread t1 = new MyThread(); Thread t2 = new MyThread(); Thread t3 = new MyThread(); Thread t4 = new MyThread(); Thread t5 = new MyThread(); //將線程放入池中進行執(zhí)行 pool.execute(t1); pool.execute(t2); pool.execute(t3); pool.execute(t4); pool.execute(t5); //關閉線程池 pool.shutdown(); } } |
輸出結果
pool-1-thread-1正在執(zhí)行。。。
pool-1-thread-2正在執(zhí)行。。。
pool-1-thread-1正在執(zhí)行。。。
pool-1-thread-2正在執(zhí)行。。。
pool-1-thread-1正在執(zhí)行。。。
3: newCachedThreadPool
TestCachedThreadPool.java
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
publicclass TestCachedThreadPool { publicstaticvoid main(String[] args) { //創(chuàng)建一個可重用固定線程數(shù)的線程池 ExecutorService pool = Executors.newCachedThreadPool(); //創(chuàng)建實現(xiàn)了Runnable接口對象，Thread對象當然也實現(xiàn)了Runnable接口 Thread t1 = new MyThread(); Thread t2 = new MyThread(); Thread t3 = new MyThread(); Thread t4 = new MyThread(); Thread t5 = new MyThread(); //將線程放入池中進行執(zhí)行 pool.execute(t1); pool.execute(t2); pool.execute(t3); pool.execute(t4); pool.execute(t5); //關閉線程池 pool.shutdown(); }} |
輸出結果：
pool-1-thread-2正在執(zhí)行。。。
pool-1-thread-4正在執(zhí)行。。。
pool-1-thread-3正在執(zhí)行。。。
pool-1-thread-1正在執(zhí)行。。。
pool-1-thread-5正在執(zhí)行。。。
4 newScheduledThreadPool
TestScheduledThreadPoolExecutor.java
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
|
publicclass TestScheduledThreadPoolExecutor { publicstaticvoid main(String[] args) { ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(1); exec.scheduleAtFixedRate(new Runnable() {//每隔一段時間就觸發(fā)異常 @Override publicvoid run() { //throw new RuntimeException(); System.out.println("================"); } }, 1000, 5000, TimeUnit.MILLISECONDS); exec.scheduleAtFixedRate(new Runnable() {//每隔一段時間打印系統(tǒng)時間，證明兩者是互不影響的 @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 - 池中所保存的線程數(shù)，包括空閑線程。
- maximumPoolSize-池中允許的最大線程數(shù)。
- keepAliveTime - 當線程數(shù)大于核心時，此為終止前多余的空閑線程等待新任務的最長時間。
- unit - keepAliveTime 參數(shù)的時間單位。
- workQueue - 執(zhí)行前用于保持任務的隊列。此隊列僅保持由 execute方法提交的 Runnable任務。
- threadFactory - 執(zhí)行程序創(chuàng)建新線程時使用的工廠。
- handler - 由于超出線程范圍和隊列容量而使執(zhí)行被阻塞時所使用的處理程序。
- ThreadPoolExecutor是Executors類的底層實現(xiàn)。
在JDK幫助文檔中，有如此一段話：
“強烈建議程序員使用較為方便的Executors工廠方法Executors.newCachedThreadPool()（無界線程池，可以進行自動線程回收）、Executors.newFixedThreadPool(int)（固定大小線程池）Executors.newSingleThreadExecutor()（單個后臺線程）
它們均為大多數(shù)使用場景預定義了設置。”
下面介紹一下幾個類的源碼：
ExecutorService newFixedThreadPool (int nThreads):固定大小線程池。
可以看到，corePoolSize和maximumPoolSize的大小是一樣的（實際上，后面會介紹，如果使用無界queue的話maximumPoolSize參數(shù)是沒有意義的），keepAliveTime和unit的設值表名什么？-就是該實現(xiàn)不想keep alive！最后的BlockingQueue選擇了LinkedBlockingQueue，該queue有一個特點，他是無界的。
|
1
2
3
4
5
6
7
8
9
|
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); } |
ExecutorService newSingleThreadExecutor()：單線程
|
1
2
3
4
5
6
7
8
9
10
11
|
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); } |
ExecutorService newCachedThreadPool()：無界線程池，可以進行自動線程回收
這個實現(xiàn)就有意思了。首先是無界的線程池，所以我們可以發(fā)現(xiàn)maximumPoolSize為big big。其次BlockingQueue的選擇上使用SynchronousQueue?？赡軐τ谠揃lockingQueue有些陌生，簡單說：該QUEUE中，每個插入操作必須等待另一個線程的對應移除操作。
|
1
2
3
4
5
6
7
8
9
|
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); } |
先從BlockingQueue<Runnable> workQueue這個入?yún)㈤_始說起。在JDK中，其實已經(jīng)說得很清楚了，一共有三種類型的queue。
所有BlockingQueue 都可用于傳輸和保持提交的任務?？梢允褂么岁犃信c池大小進行交互：
如果運行的線程少于 corePoolSize，則 Executor始終首選添加新的線程，而不進行排隊。（如果當前運行的線程小于corePoolSize，則任務根本不會存放，添加到queue中，而是直接抄家伙（thread）開始運行）
如果運行的線程等于或多于 corePoolSize，則 Executor始終首選將請求加入隊列，而不添加新的線程。
如果無法將請求加入隊列，則創(chuàng)建新的線程，除非創(chuàng)建此線程超出 maximumPoolSize，在這種情況下，任務將被拒絕。
queue上的三種類型。
排隊有三種通用策略：
- 直接提交。工作隊列的默認選項是 SynchronousQueue，它將任務直接提交給線程而不保持它們。在此，如果不存在可用于立即運行任務的線程，則試圖把任務加入隊列將失敗，因此會構造一個新的線程。此策略可以避免在處理可能具有內部依賴性的請求集時出現(xiàn)鎖。直接提交通常要求無界maximumPoolSizes 以避免拒絕新提交的任務。當命令以超過隊列所能處理的平均數(shù)連續(xù)到達時，此策略允許無界線程具有增長的可能性。
- 無界隊列。使用無界隊列（例如，不具有預定義容量的 LinkedBlockingQueue）將導致在所有 corePoolSize 線程都忙時新任務在隊列中等待。這樣，創(chuàng)建的線程就不會超過 corePoolSize。（因此，maximumPoolSize的值也就無效了。）當每個任務完全獨立于其他任務，即任務執(zhí)行互不影響時，適合于使用無界隊列；例如，在 Web頁服務器中。這種排隊可用于處理瞬態(tài)突發(fā)請求，當命令以超過隊列所能處理的平均數(shù)連續(xù)到達時，此策略允許無界線程具有增長的可能性。
- 有界隊列。當使用有限的 maximumPoolSizes時，有界隊列（如 ArrayBlockingQueue）有助于防止資源耗盡，但是可能較難調整和控制。隊列大小和最大池大小可能需要相互折衷：使用大型隊列和小型池可以最大限度地降低 CPU 使用率、操作系統(tǒng)資源和上下文切換開銷，但是可能導致人工降低吞吐量。如果任務頻繁阻塞（例如，如果它們是 I/O邊界），則系統(tǒng)可能為超過您許可的更多線程安排時間。使用小型隊列通常要求較大的池大小，CPU使用率較高，但是可能遇到不可接受的調度開銷，這樣也會降低吞吐量。
BlockingQueue的選擇。
例子一：使用直接提交策略，也即SynchronousQueue。
首先SynchronousQueue是無界的，也就是說他存數(shù)任務的能力是沒有限制的，但是由于該Queue本身的特性，在某次添加元素后必須等待其他線程取走后才能繼續(xù)添加。在這里不是核心線程便是新創(chuàng)建的線程，但是我們試想一樣下，下面的場景。
我們使用一下參數(shù)構造ThreadPoolExecutor：
|
1
2
3
4
5
6
7
8
9
|
new ThreadPoolExecutor( 2, 3, 30, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new RecorderThreadFactory("CookieRecorderPool"), new ThreadPoolExecutor.CallerRunsPolicy()); |
當核心線程已經(jīng)有2個正在運行.
- 此時繼續(xù)來了一個任務（A），根據(jù)前面介紹的“如果運行的線程等于或多于 corePoolSize，則 Executor始終首選將請求加入隊列，而不添加新的線程。”,所以A被添加到queue中。
- 又來了一個任務（B），且核心2個線程還沒有忙完，OK，接下來首先嘗試1中描述，但是由于使用的SynchronousQueue，所以一定無法加入進去。
- 此時便滿足了上面提到的“如果無法將請求加入隊列，則創(chuàng)建新的線程，除非創(chuàng)建此線程超出maximumPoolSize，在這種情況下，任務將被拒絕。”，所以必然會新建一個線程來運行這個任務。
- 暫時還可以，但是如果這三個任務都還沒完成，連續(xù)來了兩個任務，第一個添加入queue中，后一個呢？queue中無法插入，而線程數(shù)達到了maximumPoolSize，所以只好執(zhí)行異常策略了。
所以在使用SynchronousQueue通常要求maximumPoolSize是無界的，這樣就可以避免上述情況發(fā)生（如果希望限制就直接使用有界隊列）。對于使用SynchronousQueue的作用jdk中寫的很清楚：此策略可以避免在處理可能具有內部依賴性的請求集時出現(xiàn)鎖。
什么意思？如果你的任務A1，A2有內部關聯(lián)，A1需要先運行，那么先提交A1，再提交A2，當使用SynchronousQueue我們可以保證，A1必定先被執(zhí)行，在A1么有被執(zhí)行前，A2不可能添加入queue中。
例子二：使用無界隊列策略，即LinkedBlockingQueue
這個就拿newFixedThreadPool來說，根據(jù)前文提到的規(guī)則：
如果運行的線程少于 corePoolSize，則 Executor 始終首選添加新的線程，而不進行排隊。那么當任務繼續(xù)增加，會發(fā)生什么呢？
如果運行的線程等于或多于 corePoolSize，則 Executor 始終首選將請求加入隊列，而不添加新的線程。OK，此時任務變加入隊列之中了，那什么時候才會添加新線程呢？
如果無法將請求加入隊列，則創(chuàng)建新的線程，除非創(chuàng)建此線程超出 maximumPoolSize，在這種情況下，任務將被拒絕。這里就很有意思了，可能會出現(xiàn)無法加入隊列嗎？不像SynchronousQueue那樣有其自身的特點，對于無界隊列來說，總是可以加入的（資源耗盡，當然另當別論）。換句說，永遠也不會觸發(fā)產(chǎn)生新的線程！corePoolSize大小的線程數(shù)會一直運行，忙完當前的，就從隊列中拿任務開始運行。所以要防止任務瘋長，比如任務運行的實行比較長，而添加任務的速度遠遠超過處理任務的時間，而且還不斷增加，不一會兒就爆了。
例子三：有界隊列，使用ArrayBlockingQueue。
這個是最為復雜的使用，所以JDK不推薦使用也有些道理。與上面的相比，最大的特點便是可以防止資源耗盡的情況發(fā)生。
舉例來說，請看如下構造方法：
|
1
2
3
4
5
6
7
8
9
|
new ThreadPoolExecutor( 2, 4, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2), new RecorderThreadFactory("CookieRecorderPool"), new ThreadPoolExecutor.CallerRunsPolicy()); |
假設，所有的任務都永遠無法執(zhí)行完。
對于首先來的A,B來說直接運行，接下來，如果來了C,D，他們會被放到queue中，如果接下來再來E,F，則增加線程運行E，F(xiàn)。但是如果再來任務，隊列無法再接受了，線程數(shù)也到達最大的限制了，所以就會使用拒絕策略來處理。
keepAliveTime
jdk中的解釋是：當線程數(shù)大于核心時，此為終止前多余的空閑線程等待新任務的最長時間。
有點拗口，其實這個不難理解，在使用了“池”的應用中，大多都有類似的參數(shù)需要配置。比如數(shù)據(jù)庫連接池，DBCP中的maxIdle，minIdle參數(shù)。
什么意思？接著上面的解釋，后來向老板派來的工人始終是“借來的”，俗話說“有借就有還”，但這里的問題就是什么時候還了，如果借來的工人剛完成一個任務就還回去，后來發(fā)現(xiàn)任務還有，那豈不是又要去借？這一來一往，老板肯定頭也大死了。
合理的策略：既然借了，那就多借一會兒。直到“某一段”時間后，發(fā)現(xiàn)再也用不到這些工人時，便可以還回去了。這里的某一段時間便是keepAliveTime的含義，TimeUnit為keepAliveTime值的度量。
RejectedExecutionHandler
另一種情況便是，即使向老板借了工人，但是任務還是繼續(xù)過來，還是忙不過來，這時整個隊伍只好拒絕接受了。
RejectedExecutionHandler接口提供了對于拒絕任務的處理的自定方法的機會。在ThreadPoolExecutor中已經(jīng)默認包含了4中策略，因為源碼非常簡單，這里直接貼出來。
CallerRunsPolicy：
線程調用運行該任務的 execute 本身。此策略提供簡單的反饋控制機制，能夠減緩新任務的提交速度。
|
1
2
3
4
5
6
7
8
9
|
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { r.run(); } } |
這個策略顯然不想放棄執(zhí)行任務。但是由于池中已經(jīng)沒有任何資源了，那么就直接使用調用該execute的線程本身來執(zhí)行。
AbortPolicy：
處理程序遭到拒絕將拋出運行時RejectedExecutionException
|
1
2
3
4
5
|
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { throw new RejectedExecutionException(); } |
這種策略直接拋出異常，丟棄任務。
DiscardPolicy：
不能執(zhí)行的任務將被刪除
|
1
2
3
|
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { } |
這種策略和AbortPolicy幾乎一樣，也是丟棄任務，只不過他不拋出異常。
DiscardOldestPolicy：
如果執(zhí)行程序尚未關閉，則位于工作隊列頭部的任務將被刪除，然后重試執(zhí)行程序（如果再次失敗，則重復此過程）
|
1
2
3
4
5
6
7
8
9
10
11
|
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { e.getQueue().poll(); e.execute(r); } } |
該策略就稍微復雜一些，在pool沒有關閉的前提下首先丟掉緩存在隊列中的最早的任務，然后重新嘗試運行該任務。這個策略需要適當小心。
設想:如果其他線程都還在運行，那么新來任務踢掉舊任務，緩存在queue中，再來一個任務又會踢掉queue中最老任務。
總結：
keepAliveTime和maximumPoolSize及BlockingQueue的類型均有關系。如果BlockingQueue是無界的，那么永遠不會觸發(fā)maximumPoolSize，自然keepAliveTime也就沒有了意義。
反之，如果核心數(shù)較小，有界BlockingQueue數(shù)值又較小，同時keepAliveTime又設的很小，如果任務頻繁，那么系統(tǒng)就會頻繁的申請回收線程。
|
1
2
3
4
5
6
7
|
return new ThreadPoolExecutor(nThreads, nThreads,public static ExecutorService newFixedThreadPool(int nThreads) { 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());} |
以上就是深入理解Java 線程池的詳細內容，更多關于Java 線程池的資料請關注服務器之家其它相關文章！
原文鏈接：https://www.cnblogs.com/mustone/p/5798749.html
