一文帶你弄懂Java中線程池的原理
在工作中,我們經常使用線程池,但是你真的了解線程池的原理嗎?同時,線程池工作原理和底層實現(xiàn)原理也是面試經常問的考題,所以,今天我們一起聊聊線程池的原理吧。
為什么要用線程池
使用線程池主要有以下三個原因:
- 降低資源消耗。通過重復利用已創(chuàng)建的線程降低線程創(chuàng)建和銷毀造成的消耗。
- 提升響應速度。當任務到達時,任務可以不需要等到線程創(chuàng)建就能立即執(zhí)行。
- 可以對線程做統(tǒng)一管理。線程是稀缺資源,如果無限制地創(chuàng)建,不僅會消耗系統(tǒng)資源,還會降低系統(tǒng)穩(wěn)定性,使用線程池可以進行統(tǒng)一分配、調優(yōu)和監(jiān)控。
線程池的原理
Java中的線程池頂層接口是Executor
接口,ThreadPoolExecutor
是這個接口的實現(xiàn)類。
我們先看看ThreadPoolExecutor
類。
ThreadPoolExecutor提供的構造方法
// 七個參數(shù)的構造函數(shù) public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
我們先看看這些參數(shù)是什么意思:
int corePoolSize:該線程池中核心線程數(shù)最大值
核心線程:線程池中有兩類線程,核心線程和非核心線程。核心線程默認情況下會一直存在于線程池中,即使這個核心線程什么都不干(鐵飯碗),而非核心線程如果長時間的閑置,就會被銷毀(臨時工)。
int maximumPoolSize:該線程池中線程總數(shù)最大值 。
該值等于核心線程數(shù)量 + 非核心線程數(shù)量。
long keepAliveTime:非核心線程閑置超時時長。
非核心線程如果處于閑置狀態(tài)超過該值,就會被銷毀。如果設置allowCoreThreadTimeOut(true),則會也作用于核心線程。
TimeUnit unit:keepAliveTime的單位。
TimeUnit是一個枚舉類型。
BlockingQueue workQueue:阻塞隊列,維護著等待執(zhí)行的Runnable任務對象。
常用的幾個阻塞隊列:
- LinkedBlockingQueue:鏈式阻塞隊列,底層數(shù)據(jù)結構是鏈表,默認大小是
Integer.MAX_VALUE
,也可以指定大小。 - ArrayBlockingQueue:數(shù)組阻塞隊列,底層數(shù)據(jù)結構是數(shù)組,需要指定隊列的大小。
- SynchronousQueue:同步隊列,內部容量為0,每個put操作必須等待一個take操作,反之亦然。
- DelayQueue:延遲隊列,該隊列中的元素只有當其指定的延遲時間到了,才能夠從隊列中獲取到該元素 。
ThreadFactory threadFactory
創(chuàng)建線程的工廠 ,用于批量創(chuàng)建線程,統(tǒng)一在創(chuàng)建線程時設置一些參數(shù),如是否守護線程、線程的優(yōu)先級等。如果不指定,會新建一個默認的線程工廠。
static class DefaultThreadFactory implements ThreadFactory { // 省略屬性 // 構造函數(shù) DefaultThreadFactory() { SecurityManager s = System.getSecurityManager(); group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; } // 省略 }
RejectedExecutionHandler handler
拒絕處理策略,線程數(shù)量大于最大線程數(shù)就會采用拒絕處理策略,四種拒絕處理的策略為 :
- ThreadPoolExecutor.AbortPolicy:默認拒絕處理策略,丟棄任務并拋出RejectedExecutionException異常。
- ThreadPoolExecutor.DiscardPolicy:丟棄新來的任務,但是不拋出異常。
- ThreadPoolExecutor.DiscardOldestPolicy:丟棄隊列頭部(最舊的)的任務,然后重新嘗試執(zhí)行程序(如果再次失敗,重復此過程)。
- ThreadPoolExecutor.CallerRunsPolicy:由調用線程處理該任務。
ThreadPoolExecutor的策略
線程池本身有一個調度線程,這個線程就是用于管理布控整個線程池里的各種任務和事務,例如創(chuàng)建線程、銷毀線程、任務隊列管理、線程隊列管理等等。
故線程池也有自己的狀態(tài)。ThreadPoolExecutor
類中使用了一些final int
常量變量來表示線程池的狀態(tài) ,分別為RUNNING、SHUTDOWN、STOP、TIDYING 、TERMINATED。
// runState is stored in the high-order bits private 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;
- 線程池創(chuàng)建后處于RUNNING狀態(tài)。
- 調用shutdown()方法后處于SHUTDOWN狀態(tài),線程池不能接受新的任務,清除一些空閑worker,不會等待阻塞隊列的任務完成。
- 調用shutdownNow()方法后處于STOP狀態(tài),線程池不能接受新的任務,中斷所有線程,阻塞隊列中沒有被執(zhí)行的任務全部丟棄。此時,poolsize=0,阻塞隊列的size也為0。
- 當所有的任務已終止,ctl記錄的”任務數(shù)量”為0,線程池會變?yōu)?strong>TIDYING狀態(tài)。接著會執(zhí)行terminated()函數(shù)。
- 線程池處在TIDYING狀態(tài)時,執(zhí)行完terminated()方法之后,就會由 TIDYING -> TERMINATED, 線程池被設置為TERMINATED狀態(tài)。
線程池主要的任務處理流程
處理任務的核心方法是execute
,我們看看 JDK 1.8 源碼中ThreadPoolExecutor
是如何處理線程任務的:
// JDK 1.8 public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); // 1.當前線程數(shù)小于corePoolSize,則調用addWorker創(chuàng)建核心線程執(zhí)行任務 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // 2.如果不小于corePoolSize,則將任務添加到workQueue隊列。 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 2.1 如果isRunning返回false(狀態(tài)檢查),則remove這個任務,然后執(zhí)行拒絕策略。 if (! isRunning(recheck) && remove(command)) reject(command); // 2.2 線程池處于running狀態(tài),但是沒有線程,則創(chuàng)建線程 else if (workerCountOf(recheck) == 0) addWorker(null, false); } // 3.如果放入workQueue失敗,則創(chuàng)建非核心線程執(zhí)行任務, // 如果這時創(chuàng)建非核心線程失敗(當前線程總數(shù)不小于maximumPoolSize時),就會執(zhí)行拒絕策略。 else if (!addWorker(command, false)) reject(command); }
ctl.get()
是獲取線程池狀態(tài),用int
類型表示。第二步中,入隊前進行了一次isRunning
判斷,入隊之后,又進行了一次isRunning
判斷。
為什么要二次檢查線程池的狀態(tài)?
在多線程的環(huán)境下,線程池的狀態(tài)是時刻發(fā)生變化的。很有可能剛獲取線程池狀態(tài)后線程池狀態(tài)就改變了。判斷是否將command
加入workqueue
是線程池之前的狀態(tài)。倘若沒有二次檢查,萬一線程池處于非RUNNING狀態(tài)(在多線程環(huán)境下很有可能發(fā)生),那么command
永遠不會執(zhí)行。
總結一下處理流程
- 線程總數(shù)量 < corePoolSize,無論線程是否空閑,都會新建一個核心線程執(zhí)行任務(讓核心線程數(shù)量快速達到corePoolSize,在核心線程數(shù)量 < corePoolSize時)。注意,這一步需要獲得全局鎖。
- 線程總數(shù)量 >= corePoolSize時,新來的線程任務會進入任務隊列中等待,然后空閑的核心線程會依次去緩存隊列中取任務來執(zhí)行(體現(xiàn)了線程復用)。
- 當緩存隊列滿了,說明這個時候任務已經多到爆棚,需要一些“臨時工”來執(zhí)行這些任務了。于是會創(chuàng)建非核心線程去執(zhí)行這個任務。注意,這一步需要獲得全局鎖。
- 緩存隊列滿了, 且總線程數(shù)達到了maximumPoolSize,則會采取上面提到的拒絕策略進行處理。
整個過程如圖所示:
ThreadPoolExecutor如何做到線程復用的
我們知道,一個線程在創(chuàng)建的時候會指定一個線程任務,當執(zhí)行完這個線程任務之后,線程自動銷毀。但是線程池卻可以復用線程,即一個線程執(zhí)行完線程任務后不銷毀,繼續(xù)執(zhí)行另外的線程任務。那么,線程池如何做到線程復用呢?
原來,ThreadPoolExecutor在創(chuàng)建線程時,會將線程封裝成工作線程worker,并放入工作線程組中,然后這個worker反復從阻塞隊列中拿任務去執(zhí)行。
這里的addWorker
方法是在上面提到的execute
方法里面調用的,先看看上半部分:
// ThreadPoolExecutor.addWorker方法源碼上半部分 private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { int wc = workerCountOf(c); if (wc >= CAPACITY || // 1.如果core是ture,證明需要創(chuàng)建的線程為核心線程,則先判斷當前線程是否大于核心線程 // 如果core是false,證明需要創(chuàng)建的是非核心線程,則先判斷當前線程數(shù)是否大于總線程數(shù) // 如果不小于,則返回false wc >= (core ? corePoolSize : maximumPoolSize)) return false; if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } }
上半部分主要是判斷線程數(shù)量是否超出閾值,超過了就返回false。我們繼續(xù)看下半部分:
// ThreadPoolExecutor.addWorker方法源碼下半部分 boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { // 1.創(chuàng)建一個worker對象 w = new Worker(firstTask); // 2.實例化一個Thread對象 final Thread t = w.thread; if (t != null) { // 3.線程池全局鎖 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 4.啟動這個線程 t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
創(chuàng)建worker
對象,并初始化一個Thread
對象,然后啟動這個線程對象。
我們接著看看Worker
類,僅展示部分源碼:
// Worker類部分源碼 private final class Worker extends AbstractQueuedSynchronizer implements Runnable{ final Thread thread; Runnable firstTask; Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } public void run() { runWorker(this); } //其余代碼略... }
Worker
類實現(xiàn)了Runnable
接口,所以Worker
也是一個線程任務。在構造方法中,創(chuàng)建了一個線程,線程的任務就是自己。故addWorker
方法調用addWorker方法源碼下半部分中的第4步t.start
,會觸發(fā)Worker
類的run
方法被JVM調用。
我們再看看runWorker
的邏輯:
// Worker.runWorker方法源代碼 final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; // 1.線程啟動之后,通過unlock方法釋放鎖 w.unlock(); // allow interrupts boolean completedAbruptly = true; try { // 2.Worker執(zhí)行firstTask或從workQueue中獲取任務,如果getTask方法不返回null,循環(huán)不退出 while (task != null || (task = getTask()) != null) { // 2.1進行加鎖操作,保證thread不被其他線程中斷(除非線程池被中斷) w.lock(); // If pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted. This // requires a recheck in second case to deal with // shutdownNow race while clearing interrupt // 2.2檢查線程池狀態(tài),倘若線程池處于中斷狀態(tài),當前線程將中斷。 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { // 2.3執(zhí)行beforeExecute beforeExecute(wt, task); Throwable thrown = null; try { // 2.4執(zhí)行任務 task.run(); } catch (RuntimeException x) { thrown = x; throw x; } catch (Error x) { thrown = x; throw x; } catch (Throwable x) { thrown = x; throw new Error(x); } finally { // 2.5執(zhí)行afterExecute方法 afterExecute(task, thrown); } } finally { task = null; w.completedTasks++; // 2.6解鎖操作 w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }
首先去執(zhí)行創(chuàng)建這個worker時就有的任務,當執(zhí)行完這個任務后,worker的生命周期并沒有結束,在while
循環(huán)中,worker會不斷地調用getTask
方法從阻塞隊列中獲取任務然后調用task.run()
執(zhí)行任務,從而達到復用線程的目的。只要getTask
方法不返回null
,此線程就不會退出。
當然,核心線程池中創(chuàng)建的線程想要拿到阻塞隊列中的任務,先要判斷線程池的狀態(tài),如果STOP或者TERMINATED,返回null
。
最后看看getTask
方法的實現(xiàn):
// Worker.getTask方法源碼 private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // Are workers subject to culling? // 1.allowCoreThreadTimeOut變量默認是false,核心線程即使空閑也不會被銷毀 // 如果為true,核心線程在keepAliveTime內仍空閑則會被銷毀。 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 2.如果運行線程數(shù)超過了最大線程數(shù),但是緩存隊列已經空了,這時遞減worker數(shù)量。 // 如果有設置允許線程超時或者線程數(shù)量超過了核心線程數(shù)量, // 并且線程在規(guī)定時間內均未poll到任務且隊列為空則遞減worker數(shù)量 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { // 3.如果timed為true(想想哪些情況下timed為true),則會調用workQueue的poll方法獲取任務. // 超時時間是keepAliveTime。如果超過keepAliveTime時長, // poll返回了null,上邊提到的while循序就會退出,線程也就執(zhí)行完了。 // 如果timed為false(allowCoreThreadTimeOut為false // 且wc > corePoolSize為false),則會調用workQueue的take方法阻塞在當前。 // 隊列中有任務加入時,線程被喚醒,take方法返回任務,并執(zhí)行。 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
核心線程的會一直卡在workQueue.take
方法,被阻塞并掛起,不會占用CPU資源,直到拿到Runnable
然后返回(當然如果allowCoreThreadTimeOut設置為true
,那么核心線程就會去調用poll
方法,因為poll
可能會返回null
,所以這時候核心線程滿足超時條件也會被銷毀)。
非核心線程會workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) ,如果超時還沒有拿到,下一次循環(huán)判斷compareAndDecrementWorkerCount就會返回null
,Worker對象的run()
方法循環(huán)體的判斷為null
,任務結束,然后線程被系統(tǒng)回收 。
四種常見的線程池
Executors
類中提供的幾個靜態(tài)方法來創(chuàng)建線程池。
newCachedThreadPool
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }
CacheThreadPool
的運行流程如下:
- 提交任務進線程池。
- 因為corePoolSize為0的關系,不創(chuàng)建核心線程,線程池最大為Integer.MAX_VALUE。
- 嘗試將任務添加到SynchronousQueue隊列。
- 如果SynchronousQueue入列成功,等待被當前運行的線程空閑后拉取執(zhí)行。如果當前沒有空閑線程,那么就創(chuàng)建一個非核心線程,然后從SynchronousQueue拉取任務并在當前線程執(zhí)行。
- 如果SynchronousQueue已有任務在等待,入列操作將會阻塞。
當需要執(zhí)行很多短時間的任務時,CacheThreadPool的線程復用率比較高, 會顯著的提高性能。而且線程60s后會回收,意味著即使沒有任務進來,CacheThreadPool并不會占用很多資源。
newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }
核心線程數(shù)量和總線程數(shù)量相等,都是傳入的參數(shù)nThreads,所以只能創(chuàng)建核心線程,不能創(chuàng)建非核心線程。因為LinkedBlockingQueue的默認大小是Integer.MAX_VALUE,故如果核心線程空閑,則交給核心線程處理;如果核心線程不空閑,則入列等待,直到核心線程空閑。
與CachedThreadPool的區(qū)別:
- 因為 corePoolSize == maximumPoolSize ,所以FixedThreadPool只會創(chuàng)建核心線程。 而CachedThreadPool因為corePoolSize=0,所以只會創(chuàng)建非核心線程。
- 在 getTask() 方法,如果隊列里沒有任務可取,線程會一直阻塞在 LinkedBlockingQueue.take() ,線程不會被回收。 CachedThreadPool會在60s后收回。
- 由于線程不會被回收,會一直卡在阻塞,所以沒有任務的情況下, FixedThreadPool占用資源更多。
- 都幾乎不會觸發(fā)拒絕策略,但是原理不同。FixedThreadPool是因為阻塞隊列可以很大(最大為Integer最大值),故幾乎不會觸發(fā)拒絕策略;CachedThreadPool是因為線程池很大(最大為Integer最大值),幾乎不會導致線程數(shù)量大于最大線程數(shù),故幾乎不會觸發(fā)拒絕策略。
newSingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }
有且僅有一個核心線程( corePoolSize == maximumPoolSize=1),使用了LinkedBlockingQueue(容量很大),所以,不會創(chuàng)建非核心線程。所有任務按照先來先執(zhí)行的順序執(zhí)行。如果這個唯一的線程不空閑,那么新來的任務會存儲在任務隊列里等待執(zhí)行。
newScheduledThreadPool
創(chuàng)建一個定長線程池,支持定時及周期性任務執(zhí)行。
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); } //ScheduledThreadPoolExecutor(): public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS, new DelayedWorkQueue()); }
四種常見的線程池基本夠我們使用了,但是《阿里巴巴開發(fā)手冊》不建議我們直接使用Executors類中的線程池,而是通過ThreadPoolExecutor
的方式,這樣的處理方式讓寫的同學需要更加明確線程池的運行規(guī)則,規(guī)避資源耗盡的風險。
但如果你及團隊本身對線程池非常熟悉,又確定業(yè)務規(guī)模不會大到資源耗盡的程度(比如線程數(shù)量或任務隊列長度可能達到Integer.MAX_VALUE)時,其實是可以使用JDK提供的這幾個接口的,它能讓我們的代碼具有更強的可讀性。
小結
在工作中,很多人因為不了解線程池的實現(xiàn)原理,把線程池配置錯誤,從而導致各種問題。希望你們閱讀完本文,能夠學會合理的使用線程池。
對于真正想弄懂java并發(fā)編程的小伙伴,網(wǎng)上的文章還有視頻缺乏系統(tǒng)性,我建議大家還是買點書籍看看,我推薦兩本我看過的書。
《Java并發(fā)編程實戰(zhàn)》:這本書深入淺出地介紹了Java線程和并發(fā),是一本非常棒的Java并發(fā)參考手冊。
《Java并發(fā)編程藝術》:Java并發(fā)編程的概念本來就比較復雜,我們需要的是一本能夠把原理解釋清楚的書籍,而這本《Java并發(fā)編程的藝術》書是國內作者寫的Java并發(fā)書籍,剛好就比上面那一本更簡單易懂,至少我自己看下來是這樣的感覺。
以上就是一文帶你弄懂Java中線程池的原理的詳細內容,更多關于Java線程池原理的資料請關注腳本之家其它相關文章!
相關文章
IDEA2023常用配置指南(JDK/系統(tǒng)設置等常用配置)
idea很強大,但是初次安裝默認的有很多設置并不是滿足我們開發(fā)的需要,下面這篇文章主要給大家介紹了關于IDEA2023常用配置(JDK/系統(tǒng)設置等常用配置)的相關資料,需要的朋友可以參考下2023-12-12Java?restTemplate發(fā)送get請求query參數(shù)傳遞問題解決
這篇文章主要為大家介紹了Java?restTemplate發(fā)送get請求query參數(shù)傳遞問題解決,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-11-11使用SpringCache進行緩存數(shù)據(jù)庫查詢方式
這篇文章主要介紹了使用SpringCache進行緩存數(shù)據(jù)庫查詢方式,具有很好的參考價值,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2021-10-10Spring應用中使用acutator/refresh刷新屬性不生效的問題分析及解決
在Spring應用收到/actuator/refresh的POST請求后,標注了@RefreshScope以及@ConfiguratioinProperties的bean會被Spring容器重新加載,但是,在實際應用中,并沒有按照預期被Spring容器加載,本文將討論導致這種未按預期刷新的一種原因,感興趣的朋友可以參考下2024-01-01