網站首頁 編程語言 正文
今天面試被問到線程池如何復用線程的?當場就懵掉了...于是面試完畢就趕緊打開源碼看了看,在此記錄下:
我們都知道線程池的用法,一般就是先new一個ThreadPoolExecutor對象,再調用execute(Runnable runnable)傳入我們的Runnable,剩下的交給線程池處理就行了,于是這次我就從ThreadPoolExecutor的execute方法看起:
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn't, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */ int c = ctl.get(); //1.如果workerCountOf(c)即正在運行的線程數小于核心線程數,就執行addWork if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } //2.如果線程池還在運行狀態并且把任務添加到任務隊列成功 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); //3.如果線程池不在運行狀態并且從任務隊列移除任務成功,執行線程池飽和策略(默認直接拋出異常) if (! isRunning(recheck) && remove(command)) reject(command); //4.否則如果此時運行線程數==0,就直接調用addWork方法 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //5.如果2條件不成立,繼續判斷如果addWork返回false,執行線程池飽和策略 else if (!addWorker(command, false)) reject(command); }
大致過程就是如果核心線程未滿,則直接addWorker(該方法下面會再分析);如果核心線程已滿,則嘗試將任務加進消息隊列中,并再判斷如果此時運行線程數==0則調addWorker方法,否則不做任何處理(因為運行的線程處理完自己的任務后會去消息隊列中取任務來執行,下面會分析);如果任務隊列添加任務失敗,那么直接addWorker(),如果addWorker返回false,執行飽和策略,下面我們就來看看addWorker里面做了什么
/** * @param firstTask the task the new thread should run first (or * null if none). Workers are created with an initial first task * (in method execute()) to bypass queuing when there are fewer * than corePoolSize threads (in which case we always start one), * or when the queue is full (in which case we must bypass queue). * Initially idle threads are usually created via * prestartCoreThread or to replace other dying workers. * * @param core if true use corePoolSize as bound, else * maximumPoolSize. (A boolean indicator is used here rather than a * value to ensure reads of fresh values after checking other pool * state). * @return true if successful */ 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); //1.如果正在運行的線程數大于corePoolSize 或 maximumPoolSize(core代表以核心線程數還是最大線程數為邊界),return false,表示addWorker失敗 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //2.否則將運行線程數+1,并跳出這個for循環 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 } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { //3.創建一個Worker對象,傳入我們的runnable w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { 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; }
Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker. */ public void run() { runWorker(this); } final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { //1.當firstTask不為空或getTask不為空時一直循環 while (task != null || (task = getTask()) != null) { 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 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { beforeExecute(wt, task); Throwable thrown = null; try { //2.執行任務 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 { afterExecute(task, thrown); } } finally { task = null; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }
可以看到addWorker方法主要就是先判斷正在運行線程數是否超過了最大線程數(具體根據邊界取),如果未超過則創建一個worker對象,其中firstTask是我們傳入的Runnable,當然根據上面的execute方法可知當4條件滿足時,傳入的firstTask是null,Thread是用ThreadFactory創建的線程,傳入的Runnable是Worker自己,最后開啟線程,于是執行Worker這里的run、runWorker方法,在runWorker方法里,開啟一個while循環,當firstTask不為空或getTask不為空時,執行task,下面我們接著看看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.會不會淘汰空閑線程 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; //2.return null意味著回收一個Worker即淘汰一個線程 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { //3.等待指定時間 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
可以看1、2注釋,allowCoreThreadTimeOut代表存活一定時間是否對核心線程有效(默認為false),先看它為ture的情況,此時不管是核心線程還是非核心線程在3處都會等待一定時間(就是我們傳入的線程保活時間),等待時間內如果從任務隊列取到任務,則返回執行,否則timeout為true,繼續走到2,由于(timed && timedOut)和workQueue.isEmpty()均為true,返回null,代表回收一個線程;如果allowCoreThreadTimeOut為false,代表不回收核心線程,此時如果在3處沒有取到任務,繼續執行到2處,只有當wc > corePoolSize或wc > maximumPoolSize時才會執行return null,否則一直循環,相當于該線程一直處于運行狀態,直到從任務隊列拿到新的任務
原文鏈接:https://blog.csdn.net/Chushiniudao/article/details/122095315
相關推薦
- 2023-12-09 【設計模式-3.1】結構型——外觀模式
- 2022-04-19 css中link和@import的區別詳解
- 2023-10-10 ant tree拖動排序 實現同級拖動 和 跨級拖動
- 2022-06-23 Python實現希爾排序,歸并排序和桶排序的示例代碼_python
- 2022-07-08 Python中號稱神仙的六個內置函數詳解_python
- 2022-04-08 Unity?UGUI?按鈕綁定事件的?4?種方式匯總_C#教程
- 2022-09-09 python?獲取星期字符串的實例_python
- 2022-10-27 React事件處理和表單的綁定詳解_React
- 最近更新
-
- window11 系統安裝 yarn
- 超詳細win安裝深度學習環境2025年最新版(
- Linux 中運行的top命令 怎么退出?
- MySQL 中decimal 的用法? 存儲小
- get 、set 、toString 方法的使
- @Resource和 @Autowired注解
- Java基礎操作-- 運算符,流程控制 Flo
- 1. Int 和Integer 的區別,Jav
- spring @retryable不生效的一種
- Spring Security之認證信息的處理
- Spring Security之認證過濾器
- Spring Security概述快速入門
- Spring Security之配置體系
- 【SpringBoot】SpringCache
- Spring Security之基于方法配置權
- redisson分布式鎖中waittime的設
- maven:解決release錯誤:Artif
- restTemplate使用總結
- Spring Security之安全異常處理
- MybatisPlus優雅實現加密?
- Spring ioc容器與Bean的生命周期。
- 【探索SpringCloud】服務發現-Nac
- Spring Security之基于HttpR
- Redis 底層數據結構-簡單動態字符串(SD
- arthas操作spring被代理目標對象命令
- Spring中的單例模式應用詳解
- 聊聊消息隊列,發送消息的4種方式
- bootspring第三方資源配置管理
- GIT同步修改后的遠程分支