本文從多個角度全方位解析Java線程池的前世今生;
文中用到的參考資料有《Java 併發編程實戰》、《Java 併發編程的藝術》、《Java多線程編程核心技術》
需要以上這三份文檔的朋友可以私信我“資料”免費獲取!
一、簡介
什麼是線程池
線程池是一種多線程處理形式,處理過程中將任務添加到隊列,然後在創建線程後自動啟動這些任務。
為什麼要用線程池
如果併發請求數量很多,但每個線程執行的時間很短,就會出現頻繁的創建和銷燬線程。如此一來,會大大降低系統的效率,可能頻繁創建和銷燬線程的時間、資源開銷要大於實際工作的所需。
正是由於這個問題,所以有必要引入線程池。使用 線程池的好處 有以下幾點:
- 降低資源消耗 - 通過重複利用已創建的線程降低線程創建和銷燬造成的消耗。
- 提高響應速度 - 當任務到達時,任務可以不需要等到線程創建就能立即執行。
- 提高線程的可管理性 - 線程是稀缺資源,如果無限制的創建,不僅會消耗系統資源,還會降低系統的穩定性,使用線程池可以進行統一的分配,調優和監控。但是要做到合理的利用線程池,必須對其原理了如指掌。
二、Executor 框架
Executor 框架是一個根據一組執行策略調用,調度,執行和控制的異步任務的框架,目的是提供一種將”任務提交”與”任務如何運行”分離開來的機制。
核心 API 概述
Executor 框架核心 API 如下:
- Executor - 運行任務的簡單接口。
- ExecutorService - 擴展了 Executor 接口。擴展能力:支持有返回值的線程;支持管理線程的生命週期。
- ScheduledExecutorService - 擴展了 ExecutorService 接口。擴展能力:支持定期執行任務。
- AbstractExecutorService - ExecutorService 接口的默認實現。
- ThreadPoolExecutor - Executor 框架最核心的類,它繼承了 AbstractExecutorService 類。
- ScheduledThreadPoolExecutor - ScheduledExecutorService 接口的實現,一個可定時調度任務的線程池。
- Executors - 可以通過調用 Executors 的靜態工廠方法來創建線程池並返回一個 ExecutorService 對象。
Executor
Executor 接口中只定義了一個 execute 方法,用於接收一個 Runnable 對象。
<code>public interface Executor { void execute(Runnable command);}/<code>
ExecutorService
ExecutorService 接口繼承了 Executor 接口,它還提供了 invokeAll、invokeAny、shutdown、submit 等方法。
<code>public interface ExecutorService extends Executor { void shutdown(); List<runnable> shutdownNow(); boolean isShutdown(); boolean isTerminated(); boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;Future /<runnable>/<code>submit(Callable task); Future submit(Runnable task, T result); Future> submit(Runnable task); List<future>> invokeAll(Collection extends Callable > tasks) throws InterruptedException; /<future>List<future>> invokeAll(Collection extends Callable > tasks, long timeout, TimeUnit unit) throws InterruptedException; /<future>T invokeAny(Collection extends Callable > tasks) throws InterruptedException, ExecutionException; T invokeAny(Collection extends Callable > tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;}
從其支持的方法定義,不難看出:相比於 Executor 接口,ExecutorService 接口主要的擴展是:
- 支持有返回值的線程 - sumbit、invokeAll、invokeAny 方法中都支持傳入Callable 對象。
- 支持管理線程生命週期 - shutdown、shutdownNow、isShutdown 等方法。
ScheduledExecutorService
ScheduledExecutorService 接口擴展了 ExecutorService 接口。
它除了支持前面兩個接口的所有能力以外,還支持定時調度線程。
<code>public interface ScheduledExecutorService extends ExecutorService { public ScheduledFuture> schedule(Runnable command, long delay, TimeUnit unit); publicScheduledFuture /<code>schedule(Callable callable, long delay, TimeUnit unit); public ScheduledFuture> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit); public ScheduledFuture> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);}
其擴展的接口提供以下能力:
- schedule 方法可以在指定的延時後執行一個 Runnable 或者 Callable 任務。
- scheduleAtFixedRate 方法和 scheduleWithFixedDelay 方法可以按照指定時間間隔,定期執行任務。
三、ThreadPoolExecutor
java.uitl.concurrent.ThreadPoolExecutor 類是 Executor 框架中最核心的類。所以,本文將著重講述一下這個類。
重要字段
ThreadPoolExecutor 有以下重要字段:
<code>private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));private static final int COUNT_BITS = Integer.SIZE - 3;private static final int CAPACITY = (1 << COUNT_BITS) - 1;// runState is stored in the high-order bitsprivate static final int RUNNING = -1 << COUNT_BITS;private static final int SHUTDOWN = 0 << COUNT_BITS;private static final int STOP = 1 << COUNT_BITS;private static final int TIDYING = 2 << COUNT_BITS;private static final int TERMINATED = 3 << COUNT_BITS;/<code>
參數說明:
- ctl - 用於控制線程池的運行狀態和線程池中的有效線程數量。它包含兩部分的信息:線程池的運行狀態 (runState)線程池內有效線程的數量 (workerCount)可以看到,ctl 使用了 Integer 類型來保存,高 3 位保存 runState,低 29 位保存 workerCount。COUNT_BITS 就是 29,CAPACITY 就是 1 左移 29 位減 1(29 個 1),這個常量表示 workerCount 的上限值,大約是 5 億。
- 運行狀態 - 線程池一共有五種運行狀態:RUNNING - 運行狀態。接受新任務,並且也能處理阻塞隊列中的任務。SHUTDOWN - 關閉狀態。不接受新任務,但可以處理阻塞隊列中的任務。在線程池處於 RUNNING 狀態時,調用 shutdown 方法會使線程池進入到該狀態。finalize 方法在執行過程中也會調用 shutdown 方法進入該狀態。STOP - 停止狀態。不接受新任務,也不處理隊列中的任務。會中斷正在處理任務的線程。在線程池處於 RUNNING 或 SHUTDOWN 狀態時,調用 shutdownNow 方法會使線程池進入到該狀態。TIDYING - 整理狀態 。如果所有的任務都已終止了,workerCount (有效線程數) 為 0,線程池進入該狀態後會調用 terminated 方法進入 TERMINATED 狀態。TERMINATED - 已終止狀態。在 terminated 方法執行完後進入該狀態。默認 terminated 方法中什麼也沒有做。進入 TERMINATED 的條件如下:線程池不是 RUNNING 狀態;線程池狀態不是 TIDYING 狀態或 TERMINATED 狀態;如果線程池狀態是 SHUTDOWN 並且 workerQueue 為空;workerCount 為 0;設置 TIDYING 狀態成功。
構造方法
ThreadPoolExecutor 有四個構造方法,前三個都是基於第四個實現。第四個構造方法定義如下:
<code>public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {/<runnable>/<code>
參數說明:
- corePoolSize - 核心線程數量。當有新任務通過 execute 方法提交時 ,線程池會執行以下判斷:如果運行的線程數少於 corePoolSize,則創建新線程來處理任務,即使線程池中的其他線程是空閒的。如果線程池中的線程數量大於等於 corePoolSize 且小於 maximumPoolSize,則只有當 workQueue 滿時才創建新的線程去處理任務;如果設置的 corePoolSize 和 maximumPoolSize 相同,則創建的線程池的大小是固定的。這時如果有新任務提交,若 workQueue 未滿,則將請求放入 workQueue 中,等待有空閒的線程去從 workQueue 中取任務並處理;如果運行的線程數量大於等於 maximumPoolSize,這時如果 workQueue 已經滿了,則使用 handler 所指定的策略來處理任務;所以,任務提交時,判斷的順序為 corePoolSize => workQueue => maximumPoolSize。
- maximumPoolSize - 最大線程數量。如果隊列滿了,並且已創建的線程數小於最大線程數,則線程池會再創建新的線程執行任務。值得注意的是:如果使用了無界的任務隊列這個參數就沒什麼效果。
- keepAliveTime:線程保持活動的時間。當線程池中的線程數量大於 corePoolSize 的時候,如果這時沒有新的任務提交,核心線程外的線程不會立即銷燬,而是會等待,直到等待的時間超過了 keepAliveTime。所以,如果任務很多,並且每個任務執行的時間比較短,可以調大這個時間,提高線程的利用率。
- unit - keepAliveTime 的時間單位。有 7 種取值。可選的單位有天(DAYS),小時(HOURS),分鐘(MINUTES),毫秒(MILLISECONDS),微秒(MICROSECONDS, 千分之一毫秒)和毫微秒(NANOSECONDS, 千分之一微秒)。
- workQueue - 等待執行的任務隊列。用於保存等待執行的任務的阻塞隊列。 可以選擇以下幾個阻塞隊列。ArrayBlockingQueue - 有界阻塞隊列。此隊列是基於數組的先進先出隊列(FIFO)。此隊列創建時必須指定大小。LinkedBlockingQueue - 無界阻塞隊列。此隊列是基於鏈表的先進先出隊列(FIFO)。如果創建時沒有指定此隊列大小,則默認為 Integer.MAX_VALUE。吞吐量通常要高於 ArrayBlockingQueue。使用 LinkedBlockingQueue 意味著: maximumPoolSize 將不起作用,線程池能創建的最大線程數為 corePoolSize,因為任務等待隊列是無界隊列。Executors.newFixedThreadPool 使用了這個隊列。SynchronousQueue - 不會保存提交的任務,而是將直接新建一個線程來執行新來的任務。每個插入操作必須等到另一個線程調用移除操作,否則插入操作一直處於阻塞狀態。吞吐量通常要高於 LinkedBlockingQueue。Executors.newCachedThreadPool 使用了這個隊列。PriorityBlockingQueue - 具有優先級的無界阻塞隊列。
- threadFactory - 線程工廠。可以通過線程工廠給每個創建出來的線程設置更有意義的名字。
- handler - 飽和策略。它是 RejectedExecutionHandler 類型的變量。當隊列和線程池都滿了,說明線程池處於飽和狀態,那麼必須採取一種策略處理提交的新任務。線程池支持以下策略:AbortPolicy - 丟棄任務並拋出異常。這也是默認策略。DiscardPolicy - 丟棄任務,但不拋出異常。DiscardOldestPolicy - 丟棄隊列最前面的任務,然後重新嘗試執行任務(重複此過程)。CallerRunsPolicy - 只用調用者所在的線程來運行任務。如果以上策略都不能滿足需要,也可以通過實現 RejectedExecutionHandler 接口來定製處理策略。如記錄日誌或持久化不能處理的任務。
execute 方法
默認情況下,創建線程池之後,線程池中是沒有線程的,需要提交任務之後才會創建線程。
提交任務可以使用 execute 方法,它是 ThreadPoolExecutor 的核心方法,通過這個方法可以向線程池提交一個任務,交由線程池去執行。
execute 方法工作流程如下:
- 如果 workerCount < corePoolSize,則創建並啟動一個線程來執行新提交的任務;
- 如果 workerCount >= corePoolSize,且線程池內的阻塞隊列未滿,則將任務添加到該阻塞隊列中;
- 如果 workerCount >= corePoolSize && workerCount < maximumPoolSize,且線程池內的阻塞隊列已滿,則創建並啟動一個線程來執行新提交的任務;
- 如果workerCount >= maximumPoolSize,並且線程池內的阻塞隊列已滿,則根據拒絕策略來處理該任務, 默認的處理方式是直接拋異常。
其他重要方法
在 ThreadPoolExecutor 類中還有一些重要的方法:
- submit - 類似於 execute,但是針對的是有返回值的線程。submit 方法是在 ExecutorService 中聲明的方法,在 AbstractExecutorService 就已經有了具體的實現。ThreadPoolExecutor 直接複用 AbstractExecutorService 的 submit 方法。
- shutdown - 不會立即終止線程池,而是要等所有任務緩存隊列中的任務都執行完後才終止,但再也不會接受新的任務。將線程池切換到 SHUTDOWN 狀態;並調用 interruptIdleWorkers 方法請求中斷所有空閒的 worker;最後調用 tryTerminate 嘗試結束線程池。
- shutdownNow - 立即終止線程池,並嘗試打斷正在執行的任務,並且清空任務緩存隊列,返回尚未執行的任務。與 shutdown 方法類似,不同的地方在於:設置狀態為 STOP;中斷所有工作線程,無論是否是空閒的;取出阻塞隊列中沒有被執行的任務並返回。
- isShutdown - 調用了 shutdown 或 shutdownNow 方法後,isShutdown 方法就會返回 true。
- isTerminaed - 當所有的任務都已關閉後,才表示線程池關閉成功,這時調用 isTerminaed 方法會返回 true。
- setCorePoolSize - 設置核心線程數大小。
- setMaximumPoolSize - 設置最大線程數大小。
- getTaskCount - 線程池已經執行的和未執行的任務總數;
- getCompletedTaskCount - 線程池已完成的任務數量,該值小於等於 taskCount;
- getLargestPoolSize - 線程池曾經創建過的最大線程數量。通過這個數據可以知道線程池是否滿過,也就是達到了 maximumPoolSize;
- getPoolSize - 線程池當前的線程數量;
- getActiveCount - 當前線程池中正在執行任務的線程數量。
使用示例
<code>public class ThreadPoolExecutorDemo { public static void main(String[] args) { ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 500, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<runnable>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy()); for (int i = 0; i < 100; i++) { threadPoolExecutor.execute(new MyThread()); String info = String.format("線程池中線程數目:%s,隊列中等待執行的任務數目:%s,已執行玩別的任務數目:%s", threadPoolExecutor.getPoolSize(), threadPoolExecutor.getQueue().size(), threadPoolExecutor.getCompletedTaskCount()); System.out.println(info); } threadPoolExecutor.shutdown(); } static class MyThread implements Runnable { @Override public void run() { System.out.println(Thread.currentThread().getName() + " 執行"); } }}/<runnable>/<code>
四、Executors
JDK 的 Executors 類中提供了幾種具有代表性的線程池,這些線程池 都是基於 ThreadPoolExecutor 的定製化實現。
在實際使用線程池的場景中,我們往往不是直接使用 ThreadPoolExecutor ,而是使用 JDK 中提供的具有代表性的線程池實例。
newSingleThreadExecutor
創建一個單線程的線程池。
只會創建唯一的工作線程來執行任務,保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行。 如果這個唯一的線程因為異常結束,那麼會有一個新的線程來替代它 。
單工作線程最大的特點是:可保證順序地執行各個任務。
示例:
<code>public class SingleThreadExecutorDemo { public static void main(String[] args) { ExecutorService executorService = Executors.newSingleThreadExecutor(); for (int i = 0; i < 100; i++) { executorService.execute(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName() + " 執行"); } }); } executorService.shutdown(); }}/<code>
newFixedThreadPool
創建一個固定大小的線程池。
每次提交一個任務就會新創建一個工作線程,如果工作線程數量達到線程池最大線程數,則將提交的任務存入到阻塞隊列中。
FixedThreadPool 是一個典型且優秀的線程池,它具有線程池提高程序效率和節省創建線程時所耗的開銷的優點。但是,在線程池空閒時,即線程池中沒有可運行任務時,它不會釋放工作線程,還會佔用一定的系統資源。
示例:
<code>public class FixedThreadPoolDemo { public static void main(String[] args) { ExecutorService executorService = Executors.newFixedThreadPool(3); for (int i = 0; i < 100; i++) { executorService.execute(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName() + " 執行"); } }); } executorService.shutdown(); }}/<code>
newCachedThreadPool
創建一個可緩存的線程池。
- 如果線程池長度超過處理任務所需要的線程數,就會回收部分空閒的線程;
- 如果長時間沒有往線程池中提交任務,即如果工作線程空閒了指定的時間(默認為 1 分鐘),則該工作線程將自動終止。終止後,如果你又提交了新的任務,則線程池重新創建一個工作線程。
- 此線程池不會對線程池大小做限制,線程池大小完全依賴於操作系統(或者說 JVM)能夠創建的最大線程大小。 因此,使用 CachedThreadPool時,一定要注意控制任務的數量,否則,由於大量線程同時運行,很有會造成系統癱瘓。
示例:
<code>public class CachedThreadPoolDemo { public static void main(String[] args) { ExecutorService executorService = Executors.newCachedThreadPool(); for (int i = 0; i < 100; i++) { executorService.execute(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName() + " 執行"); } }); } executorService.shutdown(); }}/<code>
newScheduleThreadPool
創建一個大小無限的線程池。此線程池支持定時以及週期性執行任務的需求。
<code>public class ScheduledThreadPoolDemo { public static void main(String[] args) { schedule(); scheduleAtFixedRate(); } private static void schedule() { ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5); for (int i = 0; i < 100; i++) { executorService.schedule(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName() + " 執行"); } }, 1, TimeUnit.SECONDS); } executorService.shutdown(); } private static void scheduleAtFixedRate() { ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5); for (int i = 0; i < 100; i++) { executorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName() + " 執行"); } }, 1, 1, TimeUnit.SECONDS); } executorService.shutdown(); }}/<code>
關注我,後續更多幹貨奉上!
閱讀更多 老男孩的成長之路 的文章