## 一、為什麼需要執行緒池 在實際使用中，執行緒是很佔用系統資源的，如果對執行緒管理不完善的話很容易導致系統問題。因此，在大多數併發框架中都會使用執行緒池來管理執行緒，使用執行緒池管理執行緒主要有如下好處： - 1、使用執行緒池可以重複利用已有的執行緒繼續執行任務，避免執行緒在建立和銷燬時造成的消耗 - 2、由於沒有執行緒建立和銷燬時的消耗，可以提高系統響應速度 - 3、通過執行緒可以對執行緒進行合理的管理，根據系統的承受能力調整可執行執行緒數量的大小等 ## 二、工作原理 **流程圖：**  **執行緒池執行所提交的任務過程：** ▪ 1、比如我們設定核心執行緒池的數量為30個，不管有沒有使用者連線，我們總是保證30個連線，這個就是核心執行緒數，這裡的核心執行緒數不一定是30你可以根據你的需求、業務和併發訪問量來設定，**先判斷執行緒池中核心執行緒池所有的執行緒是否都在執行任務**，如果不是，則新建立一個執行緒執行剛提交的任務，否則，核心執行緒池中所有的執行緒都在執行任務，則進入第2步； ▪ 2、如果我們核心執行緒數的30個數量已經滿了，就需要到阻塞佇列中去檢視，判斷當前阻塞佇列是否已滿，如果未滿，則將提交的任務放置在阻塞佇列中等待執行；否則，則進入第3步； ▪ 3、判斷執行緒池中所有的執行緒是否都在執行任務，如果沒有，則建立一個新的執行緒來執行任務，否則，則交給**飽和策略**進行處理，也叫**拒絕策略**，等下我們會有詳細的介紹 **注意：** 這裡有一個**核心執行緒數和一個執行緒池數量**，這兩個是不同的概念，**核心執行緒數**代表我能夠維護常用的執行緒開銷，而**執行緒池數量**則代表我最大能夠建立的執行緒數，例如在我們農村每家每戶都有吃水的井，基本上有**半井深的水**就可以維持我們的日常生活的使用，這裡的半井深的水就好比我們的**核心執行緒數**，還有一半的容量是我們井能夠容納的最大水資源了，超過了就不行，水就會漫出來，這個就類似於我們的**執行緒池的數量**，不知道這裡說明大家是否能夠更好的進行理解 ## 三、執行緒池的分類  **1.newCachedThreadPool：** 建立一個可根據需要建立新執行緒的執行緒池，但是在以前構造的執行緒可用時講重用它們，並在需要時使用提供的ThreadFactory 建立新執行緒 > **特徵：** (1) 執行緒池中的數量沒有固定，可以達到最大值（Integer.MAX_VALUE=2147483647） (2) 執行緒池中的執行緒可進行快取重複利用和回收（回收預設時間為1分鐘） (3) 當執行緒池中，沒有可用執行緒，會重新建立一個執行緒 **2.newFixedThreadPool：** 建立一個可重用固定執行緒數的執行緒池，以共享的無界佇列方式來執行這些執行緒，在任意點，在大多數nThreads執行緒會處於處理任務的活動狀態。如果在所有執行緒處於活動狀態時提交附件任務，則在有可用執行緒之前，附件任務將在佇列中等待，如果在關閉前的執行期間由於失敗而導致任何執行緒終止，那麼一個新執行緒將代替它執行後續的任務（如果需要）。在某個執行緒被顯式關閉之前，池中的執行緒將一直存在 > **特徵：** (1) 執行緒池中的執行緒處於一定的量，可以很好的控制執行緒的併發量 (2) 執行緒可以重複被使用，在顯示關閉之前，都將一直存在 (3) 超過一定量的執行緒被提交時需在佇列中等待 **3.newSingleThreadExecutor：** 建立一個使用單個 worker 執行緒的Executor ，以無界佇列方式來執行該執行緒。（注意，如果因為在關閉前的執行期間出現失敗而終止了此單個執行緒，那麼如果需要，一個新執行緒將代替它執行後續的任務）。可保證順序地執行各個任務，並且在任意給定的時間不會有多個執行緒是活動的，與其他等效的 ```newFixedThreadPool(1)```不同，可保證無需重新配置此方法所返回的執行程式即可使用其他的執行緒 > **特徵：** (1) 執行緒池中最多執行一個執行緒，之後提交的執行緒將會排在佇列中以此執行 **4.newSingleThreadScheduledExecutor：** 建立一個單執行緒執行程式，它可安排在給定延遲後執行命令或者定期執行 > **特徵：** (1) 執行緒池中最多執行一個執行緒，之後提交的執行緒活動將會排在佇列中依次執行 (2) 可定時或者延遲執行執行緒活動 **5.newScheduledThreadPool：** 建立一個執行緒池，它可安排在給定延遲後執行命令或者定期的執行 > **特徵：** (1) 執行緒池中具有執行數量的執行緒，即便是空執行緒也將保留 (2) 可定時或者延遲執行執行緒活動 **6.newWorkStealingPool：** 建立一個帶並行級別的執行緒池，並行級別決定了同一時刻最多有多少個執行緒在執行，如不傳並行級別引數，將預設為當前系統的CPU個數 我們可以在開發工具中搜索一個叫```Executors```的類，在裡面我們可以看到我們上面所有的使用方法  ## 四、執行緒池的具體實現：ThreadPoolExecutor 執行緒工具類——Task ： ```java public class Task implements Runnable{ @Override public void run() { try { //休眠1秒 Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } //輸出執行緒名 System.out.println(Thread.currentThread().getName()+"-------running"); } } ``` #### 4.1 newCachedThreadPool **原始碼實現：** ```java public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue()); } ``` **案例：** ```java public class CacheThreadPoolDemo { public static void main(String[] args) { ExecutorService executorService = Executors.newCachedThreadPool(); for (int i = 0; i < 20; i++) { //提交任務 executorService.execute(new Task()); } //啟動有序關閉，其中先前提交的任務將被執行，但不會接受任何新任務 executorService.shutdown(); } } ``` 結果輸出： 從開始到結束我們總共輸出了20個(**pool-1-thread-1到pool-1-thread-20**)執行緒 ```java pool-1-thread-2-------running pool-1-thread-6-------running pool-1-thread-1-------running pool-1-thread-3-------running pool-1-thread-5-------running pool-1-thread-4-------running pool-1-thread-7-------running pool-1-thread-11-------running pool-1-thread-9-------running pool-1-thread-10-------running pool-1-thread-17-------running pool-1-thread-15-------running pool-1-thread-18-------running pool-1-thread-16-------running pool-1-thread-8-------running pool-1-thread-20-------running pool-1-thread-13-------running pool-1-thread-19-------running pool-1-thread-14-------running pool-1-thread-12-------running ``` #### 4.2 newFixedThreadPool **原始碼實現：** ```java public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()); } ``` **案例：** ```java public class FixedThreadPoolDemo { public static void main(String[] args) { //建立執行緒池，最多允許五個執行緒執行 ExecutorService executorService = Executors.newFixedThreadPool(5); for (int i = 0; i < 20; i++) { //提交任務 executorService.execute(new Task()); } //啟動有序關閉，其中先前提交的任務將被執行，但不會接受任何新任務 executorService.shutdown(); } } ``` 輸出結果： 我們可以看到其中的執行緒是每五個（**pool-1-thread-1到pool-1-thread-5**）一執行，在當前執行的執行緒執行中，最多允許五個執行緒進行執行 ```java pool-1-thread-4-------running pool-1-thread-2-------running pool-1-thread-1-------running pool-1-thread-3-------running pool-1-thread-5-------running pool-1-thread-4-------running pool-1-thread-5-------running pool-1-thread-3-------running pool-1-thread-2-------running pool-1-thread-1-------running pool-1-thread-4-------running pool-1-thread-2-------running pool-1-thread-1-------running pool-1-thread-3-------running pool-1-thread-5-------running pool-1-thread-4-------running pool-1-thread-5-------running pool-1-thread-2-------running pool-1-thread-1-------running pool-1-thread-3-------running ``` #### 4.3 newSingleThreadExecutor **原始碼實現：** ```java public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue())); } ``` **案例：** ```java public class SingleThreadPoolDemo { public static void main(String[] args) { ExecutorService executorService = Executors.newSingleThreadExecutor(); for (int i = 0; i < 20; i++) { //提交任務 executorService.execute(new Task()); } //啟動有序關閉，其中先前提交的任務將被執行，但不會接受任何新任務 executorService.shutdown(); } } ``` 結果輸出： 我們可以看到每次都是執行緒1輸出結果 ```java pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running pool-1-thread-1-------running ``` ## 五、執行緒池的具體實現：ScheduledThreadPoolExecutor #### 5.1 newScheduledThreadPool **案例：** ```java public static void main(String[] args) { ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3); // for (int i = 0; i < 20; i++) { System.out.println(System.currentTimeMillis()); scheduledExecutorService.schedule(new Runnable() { @Override public void run() { System.out.println("延遲三秒執行"); System.out.println(System.currentTimeMillis()); } },3, TimeUnit.SECONDS); // } scheduledExecutorService.shutdown(); } } ``` 輸出結果： ```java 1606744468814 延遲三秒執行 1606744471815 ``` #### 5.2 newSingleThreadScheduledExecutor **案例：** ```java public static void main(String[] args) { ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(); scheduledExecutorService.scheduleAtFixedRate(new Runnable() { int i = 1; @Override public void run() { System.out.println(i); i++; } },0,1, TimeUnit.SECONDS); // scheduledExecutorService.shutdown(); } ``` 輸出結果： ```java 1 2 3 4 5 ``` ## 六、執行緒池的生命週期  一般來說執行緒池只有兩種狀態，一種是```Running```，一種是```TERMINATED```，圖中間的都是過渡狀態 ```Running```：能接受新提交的任務，並且也能處理阻塞佇列中的任務 ```SHUTDOWN```：關閉狀態，不再接受新提交的任務，但卻可以繼續處理阻塞佇列中已儲存的任務 ```STOP```：不能接受新任務，也不處理佇列中的任務，會中斷正在處理任務的執行緒 ```TIDYING```：如果所有的任務都已終止了，workerCount(有效執行緒數)為0.執行緒池進入該狀態後會呼叫terminated()方法進入TERMINATED狀態 ```TERMINATED```：在terminated()方法執行完成後進入該狀態，預設terminated()方法中什麼也沒有做 ## 七、執行緒池的建立 #### 7.1 Executors 原始碼 ```java public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) ``` #### 7.2 引數說明 ```corePoolSize```：核心執行緒池的大小 ```maximumPoolSize```：執行緒池能建立執行緒的最大個數 ```keepAliveTime```：空閒執行緒存活時間 ```unit```：時間單位，為keepAliveTime指定時間單位 ```workQueue```：阻塞佇列，用於儲存任務的阻塞佇列 ```threadFactory```：建立執行緒的工程類 ```handler```：飽和策略（拒絕策略） ## 八、阻塞佇列  **ArrayBlockingQueue：** >基於陣列的阻塞佇列實現，在ArrayBlockingQueue內部，維護了一個定長陣列，以便快取佇列中的資料物件，這是-個常用的阻塞佇列，除了一個定長陣列外，ArrayBlockingQueue內部還儲存著兩個整形變數，分別標識著佇列的頭部和尾部在陣列中的位置。 ArrayBlockingQueue在生產者放入資料和消費者獲取資料，都是共用同一個鎖物件，由此也意味著兩者無法真正並行執行，這點尤其不同於LinkedBlockingQueue;按照實現原理來分析，ArrayBlockingQueue完全可以採用分離鎖，從而實現生產者和消費者操作的完全並行執行。Doug Lea之所以沒這樣去做，也許是因為ArrayBlockingQueue的資料寫入和獲取操作已經足夠輕巧，以至於引入獨立的鎖機制，除了給程式碼帶來額外的複雜性外，其在效能上完全佔不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue間還有一個明顯的不同之處在於，前者在插入或刪除元素時不會產生或銷燬任何額外的物件例項，而後者則會生成一個額外的Node物件。這在長時間內需要高效併發地處理大批量資料的系統中，其對於GC的影響還是存在一定的區別。而在建立ArrayBlockingQueue時，我們還可以控制物件的內部鎖是否採用公平鎖，預設採用非公平鎖。 **LinkedBlockingQueue：** >基於連結串列的阻塞佇列，同ArrayListBlockingQueue類似，其內部也維持著一個數據緩衝佇列（該佇列由一個連結串列構成)，當生產者往佇列中放入一個數據時，佇列會從生產者手中獲取資料，並快取在佇列內部，而生產者立即返回;只有當佇列緩衝區達到最大值快取容量時( LinkedBlockingQueue可以通過建構函式指定該值），才會阻塞生產者佇列，直到消費者從佇列中消費掉─份資料，生產者執行緒會被喚醒，反之對打於消費者這端的處理也基於同樣的原理。而 LinkedBlockingQueue之所以能夠高效的處理併發資料，還因為其對於生產者端和消費者端分別採用了獨立的鎖來控制資料同步，這也意味著在高併發的情況下生產者和消費者可以並行地操作佇列中的資料，以此來提高整個佇列的併發效能。 **DelayQueue：** >DelayQueue中的元素只有當其指定的延遲時間到了，才能夠從佇列中獲取到該元素。DelayQueue是一個沒有大小限制的佇列，因此往佇列中插入資料的操作（生產者)永遠不會被阻塞，而只有獲取資料的操作（消費者）才會被阻塞。 使用場景︰ DelayQueue使用場景較少，但都相當巧妙，常見的例子比如使用一個DelayQueue來管理一個超時未響應的連線佇列。 **PriorityBlockingQueue：** > 基於優先順序的阻塞佇列（優先順序的判斷通過建構函式傳入的Compator物件來決定），但需要注意的是 PriorityBlockingQueue並不會阻塞資料生產者，而只會在沒有可消費的資料時，阻塞資料的消費者。因此使用的時候要特別注意，生產者生產資料的速度絕對不能快於消費者消費資料的速度，否則時間一長，會最終耗盡所有的可用堆記憶體空間。在實現PriorityBlockingQueue時，內部控制執行緒同步的鎖採用的是公平鎖。 **SynchronousQueue：** >一種無緩衝的等待佇列，類似於無中介的直接交易，有點像原始社會中的生產者和消費者，生產者拿著產品去集市銷售給產品的最終消費者，而消費者必須親自去集市找到所要商品的直接生產者，如果一方沒有找到合適的目標，那麼對不起，大家都在集市等待。相對於有緩衝的BlockingQueue來說，少了一箇中間經銷商的環節（緩衝區），如果有經銷商，生產者直接把產品批發給經銷商，而無需在意經銷商最終會將這些產品賣給那些消費者，由於經銷商可以庫存一部分商品，因此相對於直接交易模式，總體來說採用中間經銷商的模式會吞吐量高一些(可以批量買賣)﹔但另一方面，又因為經銷商的引入，使得產品從生產者到消費者中間增加了額外的交易環節，單個產品的及時響應效能可能會降低。 宣告一個SynchronousQueue有兩種不同的方式，它們之間有著不太一樣的行為。公平模式和非公平模式的區別:如果採用公平模式:SynchronousQueue會採用公平鎖，並配合一個FIFO佇列來阻塞多餘的生產者和消費者，從而體系整體的公平策略; 但如果是非公平模式 ( SynchronousQueue預設) : SynchronousQueue採用非公平鎖，同時配合一個LIFO佇列來管理多餘的生產者和消費者，而後一種模式，如果生產者和消費者的處理速度有差距，則很容易出現飢渴的情況，即可能有某些生產者或者是消費者的資料永遠都得不到處理。 **注意：** **arrayblockingqueue**和**linkedblockqueue**的區別:1.佇列中鎖的實現不同 1、ArrayBlockingQueue實現的佇列中的鎖是沒有分離的，即生產和消費用的是同一個鎖; LinkedBlockingQueue實現的佇列中的鎖是分離的，即生產用的是putLock，消費是takeLock2.佇列大小初始化方式不同 2、ArrayBlockingQueue實現的佇列中必須指定佇列的大小; LinkedBlockingQueue實現的佇列中可以不指定佇列的大小，但是預設是Integer.MAX_VALUE ## 九、拒絕策略 **ThreadPoolExecutor.AbortPolicy（系統預設）：** 丟棄任務並丟擲RejectedExecutionException異常，讓你感知到任務被拒絕了，我們可以根據業務邏輯選擇重試或者放棄提交等策略 **ThreadPoolExecutor.DiscardPolicy：** 也是丟棄任務，但是不丟擲異常，相對而言存在一定的風險，因為我們提交的時候根本不知道這個任務會被丟棄，可能造成資料丟失。 **ThreadPoolExecutor.DiscardOldestPolicy：** 丟棄佇列最前面的任務，然後重新嘗試執行任務（重複此過程），通常是存活時間最長的任務，它也存在一定的資料丟失風險 **ThreadPoolExecutor.CallerRunsPolicy：** 由呼叫執行緒處理該任務 ## 十、execute()和submit()方法 #### 10.1 execute方法執行邏輯  - 如果當前執行的執行緒少於corePoolSize，則會建立新的執行緒來執行新的任務； - 如果執行的執行緒個數等於或者大於corePoolSize，則會將提交的任務存放到阻塞佇列workQueue中； - 如果當前workQueue佇列已滿的話，則會建立新的執行緒來執行任務； - 如果執行緒個數已經超過了maximumPoolSize，則會使用飽和策略RejectedExecutionHandler來進行處理 #### 10.2 Submit submit是基方法Executor.execute(Runnable)的延伸，通過建立並返回一個Future類物件可用於取消執行和/或等待完成。  ## 十一、執行緒池的關閉 - 關閉執行緒池，可以通過shutdown和shutdownNow兩個方法 - 原理：遍歷執行緒池中的所有執行緒，然後依次中斷 - 1、shutdownNow首先將執行緒池的狀態設定為STOP,然後嘗試停止所有的正在執行和未執行任務的執行緒，並返回等待執行任務的列表； - 2、shutdown只是將執行緒池的狀態設定為SHUTDOWN狀態，然後中斷所有沒有正在執行任務的執行緒