[TOC] ### Executor Executor 管理多個異步任務的執行，而無需程序員顯式地管理線程的生命周期。這里的異步是指多個任務的執行互不干擾，不需要進行同步操作。 主要有三種 Executor： * CachedThreadPool：一個任務創建一個線程； * FixedThreadPool：所有任務只能使用固定大小的線程； * SingleThreadExecutor：相當于大小為 1 的 FixedThreadPool。 ~~~java public static void main(String[] args) { ExecutorService executorService = Executors.newCachedThreadPool(); for (int i = 0; i < 5; i++) { executorService.execute(new MyRunnable()); } executorService.shutdown(); } ~~~ **為什么引入Executor線程池框架？** new Thread() 的缺點 * 每次 new Thread() 耗費性能 * 調用 new Thread() 創建的線程缺乏管理，被稱為野線程，而且可以無限制創建，之間相互競爭，會導致過多占用系統資源導致系統癱瘓。 * 不利于擴展，比如如定時執行、定期執行、線程中斷 采用線程池的優點 * 重用存在的線程，減少對象創建、消亡的開銷，性能佳 * 可有效控制最大并發線程數，提高系統資源的使用率，同時避免過多資源競爭，避免堵塞 * 提供定時執行、定期執行、單線程、并發數控制等功能 <br> ## **線程池實現原理** Java里面線程池的頂級接口是**Executor**，但是嚴格意義上講Executor并不是一個線程池，而 只是一個執行線程的工具。真正的線程池接口是**ExecutorService**。 > 蘑菇街面試，設計一個線程池 :-:  ### 并發隊列 **入隊** 非阻塞隊列：當隊列中滿了時候，放入數據，數據丟失 阻塞隊列：當隊列滿了的時候，進行等待，什么時候隊列中有出隊的數據，那么第11個再放進去 **出隊** 非阻塞隊列：如果現在隊列中沒有元素，取元素，得到的是null 阻塞隊列：等待，什么時候放進去，再取出來 線程池使用的是阻塞隊列 ### 線程池概念 線程是稀缺資源，如果被無限制的創建，不僅會消耗系統資源，還會降低系統的穩定性，合理的使用線程池對線程進行統一分配、調優和監控，有以下好處： 1. 降低資源消耗； 2. 提高響應速度； 3. 提高線程的可管理性。 Java1.5 中引入的 Executor 框架把任務的提交和執行進行解耦，只需要定義好任務，然后提交給線程池，而不用關心該任務是如何執行、被哪個線程執行，以及什么時候執行。 ### Executor類圖 [](https://github.com/frank-lam/fullstack-tutorial/blob/master/notes/JavaArchitecture/assets/820628cf179f4952812da4e8ca5de672.png) ### 線程池工作原理 線程池中的核心線程數，當提交一個任務時，線程池創建一個新線程執行任務，直到當前線程數等于corePoolSize；如果當前線程數為 corePoolSize，繼續提交的任務被保存到阻塞隊列中，等待被執行；如果阻塞隊列滿了，那就創建新的線程執行當前任務；直到線程池中的線程數達到 maxPoolSize，這時再有任務來，只能執行 reject() 處理該任務。 ## 常用的幾個線程池 ### 初始化線程池 **newFixedThreadPool()** 說明：**初始化一個指定線程數的線程池**，其中 corePoolSize == maxiPoolSize，使用 **LinkedBlockingQuene** 作為阻塞隊列 特點：即使當線程池沒有可執行任務時，也不會釋放線程。 **newCachedThreadPool()** 說明：**初始化一個可以緩存線程的線程池**，默認緩存60s，線程池的線程數可達到 Integer.MAX\_VALUE，即 2147483647，內部使用 SynchronousQueue 作為阻塞隊列； 特點：在沒有任務執行時，當線程的空閑時間超過 keepAliveTime，會自動釋放線程資源；當提交新任務時，如果沒有空閑線程，則創建新線程執行任務，會導致一定的系統開銷； 因此，使用時要注意控制并發的任務數，防止因創建大量的線程導致而降低性能。 **newSingleThreadExecutor()** 說明：**初始化只有一個線程的線程池**，內部使用 LinkedBlockingQueue 作為阻塞隊列。 特點：如果該線程異常結束，會重新創建一個新的線程繼續執行任務，唯一的線程可以保證所提交任務的順序執行 **newScheduledThreadPool()** 特點：初始化的線程池可以在指定的時間內周期性的執行所提交的任務，在實際的業務場景中可以使用該線程池定期的同步數據。 ### FixThreadPool 固定線程池 FixThreadPool ：可重用固定線程數的線程池。 ~~~ public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) { return new ThreadPoolExecutor( nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory); } ~~~ **執行機制 ：** * 若當前運行的線程數小于 `corePoolSize`，來新任務時，就創建新的線程來執行任務； * 當前運行的線程數等于 `corePoolSize` 后，如果再來新任務的話，會將任務加到 `LinkedBlockingQueue`； * 線程池中的線程執行完手頭的工作后，會在循環中反復從 `LinkedBlockingQueue` 中獲取任務來執行。 FixThreadPool 使用的是無界隊列 `LinkedBlockingQueue`（隊列容量為 Integer.MAX\_VALUE），而它會給線程池帶來如下**影響 ：** * 當線程池中的線程數達到 `corePoolSize` 后，新任務將在無界隊列中等待，因此線程池中的線程數不會超過 `corePoolSize`； * 由于使用的是一個無界隊列，所以 `maximumPoolSize` 將是一個無效參數，因為不可能存在任務隊列滿的情況，所以 FixedThreadPool 的 `corePoolSize`、`maximumPoolSize` 被設置為同一個值，且 `keepAliveTime` 將是一個無效參數； * 運行中的 FixedThreadPool（指未執行 `shutdown()` 或 `shutdownNow()` 的）不會拒絕任務，因此在任務較多的時候可能會導致 OOM。 ### SingleThreadExecutor 單一線程池 SingleThreadExecutor 是只有一個線程的線程池。 ~~~ public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor( 1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory)); } ~~~ 除了池中只有一個線程外，其他和 FixThreadPool 是基本一致的。 ### CachedThreadPool 緩存線程池 CachedThreadPool 是一個會根據需要創建新線程的線程池，但會在先前構建的線程可用時重用它。 ~~~ public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) { return new ThreadPoolExecutor( 0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), threadFactory); } ~~~ 其 `corePoolSize` 被設置為 0，`maximumPoolSize` 被設置為 Integer.MAX.VALUE，也就是無界的。雖然是無界，但由于該線程池還存在一個銷毀機制，即如果一個線程 60 秒內未被使用過，則該線程就會被銷毀，這樣就節省了很多資源。 但是，如果主線程提交任務的速度高于 `maximunPool` 中線程處理任務的速度，CachedThreadPool 將會源源不斷地創建新的線程，從而依然可能導致 CPU 耗盡或內存溢出。 執行機制 ： * 首先執行 `offer` 操作，提交任務到任務隊列。若當前 maximumPool 中有空閑線程正在執行 `poll` 操作，且主線程的 `offer` 與空閑線程的 `poll` 配對成功時，主線程將把任務交給空閑線程執行，此時視作 `execute()` 方法執行完成；否則，將執行下面的步驟。 * 當初始 `maximum` 為空，或 `maximumPool` 中沒有空閑線程時，將沒有線程執行 `poll` 操作。此時，CachedThreadPool 會創建新線程執行任務，`execute()` 方法執行完成。 ### ScheduledThreadPool 創建一個定長線程池，支持定時及周期性任務執行。延遲執行示例代碼如下： ~~~ ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5); scheduledThreadPool.schedule(new Runnable() { @Override public void run() { System.out.println("delay 3 seconds"); } }, 3, TimeUnit.SECONDS); ~~~ 表示延遲3秒執行。 ~~~ scheduledThreadPool.scheduleAtFixedRate(new Runnable() { @Override public void run() { System.out.println("delay 1 seconds, and excute every 3 seconds"); } }, 1, 3, TimeUnit.SECONDS); ~~~ 表示延遲1秒后每3秒執行一次。 ScheduledExecutorService比Timer更安全，功能更強大 #### 原理  * 它接收SchduledFutureTask類型的任務，有兩種提交任務的方式：scheduledAtFixedRate, scheduledWithFixedDelay * 它采用DelayQueue存儲等待的任務 DelayQueue內部封裝了一個PriorityQueue，它會根據time的先后時間排序，若time相同則根據sequenceNumber排序； DelayQueue也是一個無界隊列； 工作線程會從DelayQueue取已經到期的任務去執行； 執行結束后重新設置任務的到期時間，再次放回DelayQueue #### 初始化方法 ~~~java // 使用Executors靜態方法進行初始化 ExecutorService service = Executors.newSingleThreadExecutor(); // 常用方法 service.execute(new Thread()); service.submit(new Thread()); service.shutDown(); service.shutDownNow(); ~~~ ### 常用方法 #### execute與submit的區別 1. 接收的參數不一樣 2. submit有返回值，而execute沒有 用到返回值的例子，比如說我有很多個做 validation 的 task，我希望所有的 task 執行完，然后每個 task 告訴我它的執行結果，是成功還是失敗，如果是失敗，原因是什么。然后我就可以把所有失敗的原因綜合起來發給調用者。 3. submit方便Exception處理 如果你在你的 task 里會拋出 checked 或者 unchecked exception，而你又希望外面的調用者能夠感知這些 exception 并做出及時的處理，那么就需要用到 submit，通過捕獲 Future.get 拋出的異常。 #### shutDown與shutDownNow的區別 當線程池調用該方法時,線程池的狀態則立刻變成 SHUTDOWN 狀態。此時，則不能再往線程池中添加任何任務，否則將會拋出 RejectedExecutionException 異常。但是，此時線程池不會立刻退出，直到添加到線程池中的任務都已經處理完成，才會退出。 ### 內部實現 ~~~java public ThreadPoolExecutor( int corePoolSize, // 核心線程數 int maximumPoolSize, // 最大線程數 long keepAliveTime, // 線程存活時間（在 corePore<*<maxPoolSize 情況下有用） TimeUnit unit, // 存活時間的時間單位 BlockingQueue<Runnable> workQueue // 阻塞隊列（用來保存等待被執行的任務） ThreadFactory threadFactory, // 線程工廠，主要用來創建線程； RejectedExecutionHandler handler // 當拒絕處理任務時的策略 ){ this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); } ~~~ 關于 workQueue 參數，有四種隊列可供選擇： * ArrayBlockingQueue：基于數組結構的有界阻塞隊列，按 FIFO 排序任務； * LinkedBlockingQuene：基于鏈表結構的阻塞隊列，按 FIFO 排序任務； * SynchronousQuene：一個不存儲元素的阻塞隊列，每個插入操作必須等到另一個線程調用移除操作，否則插入操作一直處于阻塞狀態，吞吐量通常要高于 ArrayBlockingQuene； * PriorityBlockingQuene：具有優先級的無界阻塞隊列； 關于 handler 參數，線程池的飽和策略，當阻塞隊列滿了，且沒有空閑的工作線程，如果繼續提交任務，必須采取一種策略處理該任務，線程池提供了 4 種策略： * ThreadPoolExecutor.AbortPolicy：丟棄任務并拋出RejectedExecutionException異常。 * ThreadPoolExecutor.DiscardPolicy：丟棄任務，但是不拋出異常。 * ThreadPoolExecutor.DiscardOldestPolicy：丟棄隊列最前面的任務，然后重新嘗試執行任務（重復此過程） * ThreadPoolExecutor.CallerRunsPolicy：由調用線程處理該任務 當然也可以根據應用場景實現 RejectedExecutionHandler 接口，自定義飽和策略，如記錄日志或持久化存儲不能處理的任務。 ### 線程池的狀態 ~~~java private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); ~~~ 其中 AtomicInteger 變量 ctl 的功能非常強大：利用低 29 位表示線程池中線程數，通過高 3 位表示線程池的運行狀態： * **RUNNING**：-1 << COUNT\_BITS，即高 3 位為 111，該狀態的線程池會接收新任務，并處理阻塞隊列中的任務； * **SHUTDOWN**： 0 << COUNT\_BITS，即高 3 位為 000，該狀態的線程池不會接收新任務，但會處理阻塞隊列中的任務； * **STOP**： 1 << COUNT\_BITS，即高 3 位為 001，該狀態的線程不會接收新任務，也不會處理阻塞隊列中的任務，而且會中斷正在運行的任務； * **TIDYING**： 2 << COUNT\_BITS，即高 3 位為 010，該狀態表示線程池對線程進行整理優化； * **TERMINATED**： 3 << COUNT\_BITS，即高 3 位為 011，該狀態表示線程池停止工作； ### 線程池其他常用方法 如果執行了線程池的 prestartAllCoreThreads() 方法，線程池會提前創建并啟動所有核心線程。 ThreadPoolExecutor 提供了動態調整線程池容量大小的方法：setCorePoolSize() 和 setMaximumPoolSize()。 ### 為什么推薦使用 ThreadPoolExecutor 來創建線程或者說為什么不推薦直接用Executors來創建線程池？ 規約一 ：線程資源必須通過線程池提供，不允許在應用中自行顯示創建線程。 > 使用線程池的好處是減少在創建和銷毀線程上所消耗的時間以及系統資源開銷，解決資源不足的問題。如果不使用線程池，有可能會造成系統創建大量同類線程而導致消耗完內存或者“過度切換”的問題。 規約二 ：強制線程池不允許使用 `Executors` 去創建，而是通過 `ThreadPoolExecutor` 構造函數的方式，這樣的處理方式讓寫的同學更加明確線程池的運行規則，規避資源耗盡的風險。 > Executors 返回線程池對象的弊端如下： > > `FixedThreadPool` 和 `SingleThreadExecutor` ： 允許請求的**隊列長度**為 Integer.MAX\_VALUE，可能會**堆積大量請求**，從而導致 OOM。 > > `CachedThreadPool` 和 `ScheduledThreadPool` ： 允許創建的**線程數量**為 Integer.MAX\_VALUE，可能會**創建大量線程**，從而導致 OOM。 ## 如何擬定線程池的大小？ ### 上下文切換 多線程變編程中一般線程的個數都大于 CPU 核心的個數，而一個 CPU 核心在任意時刻只能被一個線程使用。為了讓這些線程都能得到有效執行，CPU 采取的策略是為每個線程分配時間片并輪轉的形式。當一個線程的時間片用完的時候就會重新處于就緒狀態讓給其他線程使用，這個過程就屬于一次上下文切換。 概括來說就是，當前任務在執行完 CPU 時間片切換到另一個任務之前，會先保存自己的狀態，以便下次再切換回這個任務時，可以直接加載到上次的狀態。任務從保存到再加載的過程就是一次上下文切換。 上下文切換通常是計算密集型的。也就是說，它需要相當可觀的處理器時間，在每秒幾十上百次的切換中，每次切換都需要納秒量級的時間。所以，上下文切換對系統來說意味著消耗大量的 CPU 時間，事實上，可能是操作系統中時間消耗最大的操作。 > Linux 相比與其他操作系統（包括其他類 Unix 系統）有許多，其中有一項就是，其上下文切換和模式切換的時間消耗非常少。 > ### 簡單的擬定判斷 **CPU 密集型任務(N+1)：** 這種任務消耗的主要是 CPU 資源，可以將線程數設置為 N（CPU 核心數）+1，比 CPU 核心數多出來的一個線程是為了防止線程偶發的缺頁中斷，或者其它原因導致的任務暫停而帶來的影響。一旦任務暫停，CPU 就會處于空閑狀態，而在這種情況下多出來的一個線程就可以充分利用 CPU 的空閑時間。 **I/O 密集型任務(2N)：** 這種任務應用起來，系統會用大部分的時間來處理 I/O 交互，而線程在處理 I/O 的時間段內不會占用 CPU 來處理，這時就可以將 CPU 交出給其它線程使用。因此在 I/O 密集型任務的應用中，我們可以多配置一些線程，具體的計算方法是 2N。 美團技術團隊深入解析線程池原理：https://tech.meituan.com/2020/04/02/java-pooling-pratice-in-meituan.html