摘要:線程池概述什么是線程池就是將多個(gè)線程放在一個(gè)池子里面所謂池化技術(shù)然后需要線程的時(shí)候不是創(chuàng)建一個(gè)線程而是從線程池里面獲取一個(gè)可用的線程然后執(zhí)行我們的任務(wù)線程池的優(yōu)勢(shì)降低資源消耗通過重復(fù)利用已創(chuàng)建的線程降低線程創(chuàng)建和消耗提供響應(yīng)速度當(dāng)任務(wù)到達(dá)時(shí)
線程池概述
什么是線程池
就是將多個(gè)線程放在一個(gè)池子里面(所謂池化技術(shù)),然后需要線程的時(shí)候不是創(chuàng)建一個(gè)線程,而是從線程池里面獲取一個(gè)可用的線程,然后執(zhí)行我們的任務(wù).
線程池的優(yōu)勢(shì)
降低資源消耗,通過重復(fù)利用已創(chuàng)建的線程降低線程創(chuàng)建和消耗
提供響應(yīng)速度,當(dāng)任務(wù)到達(dá)時(shí),任務(wù)可以不需要等到線程創(chuàng)建就立即執(zhí)行
提高線程的可管理性,線程是稀缺資源,如果無限制的創(chuàng)建,不僅會(huì)消耗系統(tǒng)資源,還會(huì)降低系統(tǒng)的穩(wěn)定性,使用線程池可以進(jìn)行統(tǒng)一的分配、調(diào)優(yōu)和監(jiān)控.
創(chuàng)建一個(gè)線程池并提交線程任務(wù)Java線程池最核心的類是ThreadPoolExecutor,查看ThreadPoolExecutor類關(guān)系繼承圖如下:
查看Executor接口可以通過execute方法進(jìn)行提交任務(wù)
查看ExecutorService接口可以通過submit進(jìn)行提交任務(wù)
所以ThreadPoolExecutor可以使用上述兩種方式提交任務(wù)
ThreadPoolExecutor的核心內(nèi)部類為Worker,其對(duì)資源進(jìn)行了復(fù)用,減少了創(chuàng)建線程的開銷,而其他的AbortPolicy等則是RejectedExecutionHandler接口的各種拒絕策略類
當(dāng)使用線程池并且使用有界隊(duì)列的時(shí)候,如果隊(duì)列滿了,任務(wù)添加到線程池就會(huì)有問題,針對(duì)這個(gè)問題Java線程池提供了以下拒絕策略:
AbortPolicy:使用該策略時(shí),如果線程池隊(duì)列滿了,丟掉這個(gè)任務(wù)并且拋出RejectedExecutionException異常
DiscardPolicy: 如果線程池隊(duì)列滿了,會(huì)直接丟掉這個(gè)任務(wù)并且不會(huì)有任何異常
DiscardOldestPolicy: 如果線程池隊(duì)列滿了,會(huì)將最老的(即最早進(jìn)入隊(duì)列的)任務(wù)刪除掉并騰出隊(duì)列空間,再嘗試將任務(wù)加入隊(duì)列
CallerRunsPolicy:如果任務(wù)添加到線程池失敗,那么主線程會(huì)自己去執(zhí)行該任務(wù),不會(huì)去等待線程池的任務(wù)去執(zhí)行
自定義:如果以上策略不符合業(yè)務(wù)場(chǎng)景,那么可以自己定義拒絕策略,只要實(shí)現(xiàn)RejectedExecutionHandler接口,并且實(shí)現(xiàn)rejectedExecution方法就可以了
由于核心內(nèi)部類是worker,而且worker簡(jiǎn)易,先解析worker:
Worker類源碼解析 類繼承關(guān)系private final class Worker extends AbstractQueuedSynchronizer implements Runnable
可知:Worker類繼承了AQS抽象類,實(shí)現(xiàn)了Runnable接口,重寫了AQS的一些方法,對(duì)應(yīng)的Runnable接口可以創(chuàng)建線程的動(dòng)作
類屬性private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */ // 版本號(hào) private static final long serialVersionUID = 6138294804551838833L; /** Thread this worker is running in. Null if factory fails. */ // worker 所對(duì)應(yīng)的線程 final Thread thread; /** Initial task to run. Possibly null. */ // worker初始化任務(wù),默認(rèn)第一個(gè)任務(wù) Runnable firstTask; /** Per-thread task counter */ // 每個(gè)線程任務(wù)計(jì)數(shù)器,記錄已完成任務(wù)數(shù)量 volatile long completedTasks;
說明:
1.Thread類型的thread屬性用來封裝worker,對(duì)應(yīng)形成一個(gè)線程 2.Runnable類型的firstTask其表示該worker包含的runnable對(duì)象,即用戶自定義的Runnable 3.volatile修飾的long類型的completedTasks表示已完成的任務(wù)數(shù)量類構(gòu)造函數(shù)
Worker(Runnable firstTask) { // AQS的狀態(tài)設(shè)置為-1,進(jìn)行抑制中斷直到 runWorker setState(-1); // inhibit interrupts until runWorker // 初始化第一個(gè)任務(wù) this.firstTask = firstTask; // 根據(jù)當(dāng)前worker,初始化線程 this.thread = getThreadFactory().newThread(this); }
進(jìn)行構(gòu)造worker對(duì)象,初始化對(duì)應(yīng)的屬性
worker核心函數(shù)分析/** Delegates main run loop to outer runWorker */ // 重寫Runnable的run方法,并將run方法交給外部的runWorker public void run() { runWorker(this); } // Lock methods // // The value 0 represents the unlocked state. // The value 1 represents the locked state. // 是否被獨(dú)占,0表示未被獨(dú)占,1表示被獨(dú)占 protected boolean isHeldExclusively() { return getState() != 0; } // 嘗試獲取方法 protected boolean tryAcquire(int unused) { // CAS方法設(shè)置State狀態(tài)值 if (compareAndSetState(0, 1)) { // 設(shè)置獨(dú)占線程 setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } // 嘗試釋放 protected boolean tryRelease(int unused) { // 設(shè)置獨(dú)占線程為null setExclusiveOwnerThread(null); // 設(shè)置狀態(tài)為0 setState(0); return true; } // 獲取鎖 public void lock() { acquire(1); } // 嘗試獲取鎖 public boolean tryLock() { return tryAcquire(1); } // 是否鎖 public void unlock() { release(1); } // 是否被獨(dú)占 public boolean isLocked() { return isHeldExclusively(); } // 中斷線程操作 void interruptIfStarted() { Thread t; // 當(dāng)AQS狀態(tài)>=0并且worker對(duì)象的線程不為null并且該線程沒有被中斷 if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try { // 中斷線程 t.interrupt(); } catch (SecurityException ignore) { } } }ThreadPoolExecutor類的屬性
public class ThreadPoolExecutor extends AbstractExecutorService { // 線程池的控制狀態(tài)(用來表示線程池的運(yùn)行狀態(tài)--高3位和運(yùn)行的worker數(shù)量--低29位) private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 29位的偏移量 private static final int COUNT_BITS = Integer.SIZE - 3; // 最大容量 2^29-1 private static final int CAPACITY = (1 << COUNT_BITS) - 1; // runState is stored in the high-order bits // 線程運(yùn)行狀態(tài),總共5種狀態(tài),高3位表示 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; // 對(duì)ctl進(jìn)行裝箱和拆箱動(dòng)作 // 拆分運(yùn)行狀態(tài) private static int runStateOf(int c) { return c & ~CAPACITY; } // 拆分線程數(shù)量 private static int workerCountOf(int c) { return c & CAPACITY; } // 運(yùn)行狀態(tài)和線程數(shù)量組合 private static int ctlOf(int rs, int wc) { return rs | wc; } /* * Bit field accessors that don"t require unpacking ctl. * These depend on the bit layout and on workerCount being never negative. */ // 判斷當(dāng)前的運(yùn)行狀態(tài)是否在s這個(gè)標(biāo)準(zhǔn)狀態(tài)之下 private static boolean runStateLessThan(int c, int s) { return c < s; } // 判斷當(dāng)前的運(yùn)行狀態(tài)是否在s這個(gè)標(biāo)準(zhǔn)狀態(tài)之上 private static boolean runStateAtLeast(int c, int s) { return c >= s; } // 判斷是否為運(yùn)行狀態(tài) private static boolean isRunning(int c) { return c < SHUTDOWN; } /** * Attempts to CAS-increment the workerCount field of ctl. * 嘗試以CAS方式增加ctl里的workerCount字段 */ private boolean compareAndIncrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect + 1); } /** * Attempts to CAS-decrement the workerCount field of ctl. * 嘗試以CAS方式遞減ctl里的workerCount字段 */ private boolean compareAndDecrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect - 1); } /** * 遞減ctl的workcount字段,僅僅在線程突然終止時(shí)才調(diào)用(具體見processWorkerExit) * 其他遞減在getTask內(nèi)執(zhí)行 */ private void decrementWorkerCount() { do {} while (! compareAndDecrementWorkerCount(ctl.get())); } /** * 阻塞隊(duì)列:用于保存任務(wù)和移交任務(wù)給工作線程 * 不要求workQueue執(zhí)行poll()方法返回null去判斷workQueue的isEmpty() */ private final BlockingQueueworkQueue; /** * 可重入鎖:持有鎖才可以訪問workers集合和相關(guān)的記錄 * 雖然可以使用并行集,但是通常最好使用鎖;原因是序列化 * interruptIdleWorkers需避免不需要的interrupt storms,特別是shutdown期間 * 否則退出線程將同時(shí)中斷那些尚未中斷的. */ private final ReentrantLock mainLock = new ReentrantLock(); /** * 存放工作線程集合 * Set集合包含線程池中所有線程,當(dāng)持有mainLock就可以被訪問 */ private final HashSet workers = new HashSet (); /** * 終止條件 */ private final Condition termination = mainLock.newCondition(); /** * 最大線程池容量(僅在mainLock下可以訪問) */ private int largestPoolSize; /** * 已完成任務(wù)數(shù)量.(僅在工作線程終止時(shí)更新,并且持有mainLock) */ private long completedTaskCount; /* * 下方的所有用戶控制參數(shù)都被聲明為volatile,以致于操作于最新的值 * 但是不需要鎖定,因?yàn)闆]有內(nèi)部變量依賴它們?cè)谄渌僮魃贤叫薷? */ /** * 線程工廠:所有線程都是通過工廠創(chuàng)建(通過addworker) * 所有調(diào)用必須準(zhǔn)備好addworker失敗情況(如限制線程數(shù)量的策略時(shí)候), */ private volatile ThreadFactory threadFactory; /** * 在失敗時(shí)(執(zhí)行飽和或關(guān)機(jī))調(diào)用的處理程序 */ private volatile RejectedExecutionHandler handler; /** * 線程沒有任務(wù)執(zhí)行時(shí)最多保持多久時(shí)間會(huì)終止 * 線程在存在corePoolSize或allowCoreThreadTimeOut時(shí)使用此超時(shí) */ private volatile long keepAliveTime; /** * 是否運(yùn)行核心線程超時(shí)機(jī)制 */ private volatile boolean allowCoreThreadTimeOut; /** * 線程池大小 */ private volatile int corePoolSize; /** * 最大線程池大小(受限于容量) */ private volatile int maximumPoolSize; /** * 默認(rèn)拒絕執(zhí)行策略 */ private static final RejectedExecutionHandler defaultHandler = new AbortPolicy(); /** * shutdown和shutdownNow調(diào)用時(shí)所需的權(quán)限 */ private static final RuntimePermission shutdownPerm = new RuntimePermission("modifyThread"); /* 執(zhí)行finalizer時(shí)要使用的上下文 */ private final AccessControlContext acc;
著重講解下線程池的運(yùn)行狀態(tài):
1.RUNNING:接受新任務(wù)并且處理已經(jīng)進(jìn)入阻塞隊(duì)列的任務(wù)
2.SHUTDOWN:不接受新任務(wù),但是處理已經(jīng)進(jìn)入阻塞隊(duì)列的任務(wù)
3.STOP:不接受新任務(wù),不處理已經(jīng)進(jìn)入阻塞隊(duì)列的任務(wù)并且中斷正在運(yùn)行任務(wù)
4.TIDYING:所有任務(wù)都已經(jīng)終止,workerCount為0,線程轉(zhuǎn)化為TIDYING狀態(tài)并且調(diào)用terminated鉤子函數(shù)
5.terminated鉤子函數(shù)已經(jīng)運(yùn)行完成
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;
runState單調(diào)增加,不一定要命中每個(gè)狀態(tài):
RUNNING -> SHUTDOWN:調(diào)用SHUTDOWN()時(shí),可能隱式在最后調(diào)用finalize() (RUNNING or SHUTDOWN) -> STOP:調(diào)用shutdownNow() SHUTDOWN -> TIDYING:當(dāng)隊(duì)列和線程池都為空時(shí) STOP -> TIDYING:當(dāng)線程池為空時(shí) TIDYING -> TERMINATED:當(dāng)terminated()鉤子方法已經(jīng)完成ThreadPoolExecutor類的構(gòu)造函數(shù)
ThreadPoolExecutor類總共有四個(gè)構(gòu)造函數(shù),但是前面三個(gè)都是特例最終調(diào)的都是最后一個(gè),咱先解析每個(gè)構(gòu)造函數(shù)再統(tǒng)一分析好它每一個(gè)參數(shù)的意思
1.ThreadPoolExecutor(int, int, long, TimeUnit, BlockingQueue
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueworkQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }
說明:該構(gòu)造函數(shù)默認(rèn)的線程工廠及拒絕執(zhí)行策略去創(chuàng)建ThreadPoolExecutor
2.ThreadPoolExecutor(int, int, long, TimeUnit, BlockingQueue
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueworkQueue, ThreadFactory threadFactory) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler); }
說明:該構(gòu)造函數(shù)只給出默認(rèn)的拒絕執(zhí)行策略
3.ThreadPoolExecutor(int, int, long, TimeUnit, BlockingQueue
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueworkQueue, RejectedExecutionHandler handler) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler); }
說明:該構(gòu)造函數(shù)只給出默認(rèn)的線程工廠
4.ThreadPoolExecutor(int, int, long, TimeUnit, BlockingQueue
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueworkQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { // 線程池大小不能小于0 || 最大容量不能小于0 || 最大容量不能小于線程池大小 || keepAliveTime不能小于0 if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); // 初始化相應(yīng)的屬性數(shù)據(jù) this.acc = System.getSecurityManager() == null ? null : AccessController.getContext(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }
corePoolSize:線程池大小,在創(chuàng)建線程池后,默認(rèn)情況下線程池中并沒有任何線程,而是等到有任務(wù)到來后才創(chuàng)建線程去執(zhí)行任務(wù),除非調(diào)用了prestartAllCoreThreads()或者prestartCoreThread()方法,就會(huì)預(yù)創(chuàng)建線程,即在沒有任務(wù)到來之前就創(chuàng)建corePoolSize個(gè)線程或一個(gè)線程.默認(rèn)情況下,在創(chuàng)建了線程池后,線程池中的線程數(shù)為0,當(dāng)有任務(wù)來之后,就會(huì)創(chuàng)建一個(gè)線程去執(zhí)行任務(wù),當(dāng)線程池中的線程數(shù)目達(dá)到corePoolSize后,就會(huì)把到達(dá)的任務(wù)放到緩存隊(duì)列當(dāng)中
maximumPoolSize:線程池最大線程數(shù),表示線程池中最多創(chuàng)建多少個(gè)線程
keepAliveTime:表示線程沒有任務(wù)執(zhí)行時(shí)最多保持多久時(shí)間會(huì)終止.默認(rèn)情況下只有當(dāng)線程池中的線程數(shù)大于corePoolSize時(shí),KeepAliveTime才會(huì)起作用,直到線程池中的線程數(shù)不大于corePoolSize,即當(dāng)線程池中的線程數(shù)大于CorePoolSize時(shí),如果一個(gè)線程空閑的時(shí)間達(dá)到keepAliveTime則會(huì)終止,直到線程池中的線程數(shù)不超過corePoolSize.但是如果調(diào)用了allowCoreThreadTimeOut(boolean)方法,在線程池中的線程數(shù)不大于corePoolSize時(shí),keepAliveTime參數(shù)也會(huì)起作用,直到線程池中的線程數(shù)為0
unit: 參數(shù)keepAliveTime的時(shí)間單位,有7種取值,默認(rèn)為納秒
TimeUnit.DAYS; //天 TimeUnit.HOURS; //小時(shí) TimeUnit.MINUTES; //分鐘 TimeUnit.SECONDS; //秒 TimeUnit.MILLISECONDS; //毫秒 TimeUnit.MICROSECONDS; //微妙 TimeUnit.NANOSECONDS; //納秒
workQueue: 一個(gè)阻塞隊(duì)列,用來存儲(chǔ)等待執(zhí)行的任務(wù),一般有以下幾種選擇:ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue
threadFactory:線程工廠,主要用來創(chuàng)建線程
handler:拒絕執(zhí)行策略
ThreadPoolExecutor類的核心函數(shù)分析 任務(wù)提交過程1.execute方法
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * 進(jìn)行下面三步: * * 1. 如果運(yùn)行的線程小于corePoolSize,則嘗試使用用戶定義的Runnable對(duì)象創(chuàng)建一個(gè)新的線程 * 調(diào)用addWorker函數(shù)會(huì)原子性的檢查runState和workCount,通過返回false來防止在不應(yīng)該添加 * 線程時(shí)添加了線程 * * 2. 如果一個(gè)任務(wù)能夠成功入隊(duì)列,在添加一個(gè)線程時(shí)仍需進(jìn)行雙重檢查(因?yàn)榍耙淮螜z查后該線程 * 可能死亡了或進(jìn)入到此方法時(shí)線程池已經(jīng)shutdown了,所以需要再次檢查狀態(tài));如有必要當(dāng)停止時(shí) * 還需要回滾入隊(duì)列操作,或當(dāng)線程池沒有線程時(shí)需要?jiǎng)?chuàng)建一個(gè)新線程 * * 3. 如果無法入隊(duì)列,那么需要增加一個(gè)新線程,如果此操作失敗,那么就意味著線程池已經(jīng)shutdown * 或者已經(jīng)飽和了,所以拒絕任務(wù) */ // 獲取線程池控制狀態(tài) int c = ctl.get(); // worker數(shù)量小于corePoolSize if (workerCountOf(c) < corePoolSize) { // 添加worker成功則返回,不成功則再次獲取線程池控制狀態(tài) if (addWorker(command, true)) return; c = ctl.get(); } // 線程池處于RUNNING狀態(tài),將用戶自定義的Runnable對(duì)象添加進(jìn)Queue隊(duì)列 if (isRunning(c) && workQueue.offer(command)) { // 再次檢查獲取線程池控制狀態(tài) int recheck = ctl.get(); // 若此時(shí)線程池不處于RUNNING狀態(tài),將自定義任務(wù)從workQueue隊(duì)列中移除 if (! isRunning(recheck) && remove(command)) reject(command); // 拒絕執(zhí)行命令 // worker數(shù)量等于0,添加worker else if (workerCountOf(recheck) == 0) addWorker(null, false); } // 添加worker失敗則拒絕執(zhí)行命令 else if (!addWorker(command, false)) reject(command); }
說明:當(dāng)客戶端調(diào)用submit時(shí),之后會(huì)間接調(diào)用execute函數(shù),其在將來某個(gè)時(shí)間執(zhí)行給定任務(wù),execute并不會(huì)直接運(yùn)行給定任務(wù),它主要調(diào)用addWorker方法
2.addWorker方法
addWorker主要是完成以下任務(wù):
原子性增加workerCount
將用戶給定的任務(wù)封裝成一個(gè)worker,并將此worker添加進(jìn)workers集合
啟動(dòng)worker對(duì)應(yīng)的線程,并啟動(dòng)該線程運(yùn)行worker的run方法
回滾worker的創(chuàng)建動(dòng)作,即將worker從workers集合中刪除并原子性的減少workerCount
private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) {// 外層無限循環(huán) // 獲取線程池控制狀態(tài) int c = ctl.get(); // 獲取狀態(tài) int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN &&// 狀態(tài)大于等于SHUTDOWN,初始的ctl為RUNNING,小于SHUTDOWN ! (rs == SHUTDOWN &&// 狀態(tài)為SHUTDOWN firstTask == null &&// 第一個(gè)任務(wù)為null ! workQueue.isEmpty()))// worker隊(duì)列不為空 // 返回 return false; for (;;) { // worker數(shù)量 int wc = workerCountOf(c); if (wc >= CAPACITY || // worker數(shù)量大于等于最大容量 wc >= (core ? corePoolSize : maximumPoolSize))// worker數(shù)量大于等于核心線程池大小或者最大線程池大小 return false; if (compareAndIncrementWorkerCount(c))// 比較并增加worker的數(shù)量 // 跳出外層循環(huán) break retry; // 獲取線程池控制狀態(tài) c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs)// 此次的狀態(tài)與上次獲取的狀態(tài)不相同 // 跳過剩余部分,繼續(xù)循環(huán) continue retry; // else CAS failed due to workerCount change; retry inner loop } } // worker開始標(biāo)志 boolean workerStarted = false; // worker被添加標(biāo)志 boolean workerAdded = false; Worker w = null; try { // 初始化worker w = new Worker(firstTask); // 獲取worker對(duì)應(yīng)的線程 final Thread t = w.thread; if (t != null) {// 線程不為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. // 線程池運(yùn)行狀態(tài) int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || // 小于SHUTDOWN (rs == SHUTDOWN && firstTask == null)) {// 等于SHUTDOWN并且firstTask為null if (t.isAlive()) // precheck that t is startable;線程剛添加進(jìn)來,還未啟動(dòng)就存活 // 拋出線程狀態(tài)異常 throw new IllegalThreadStateException(); // worker添加到workers集合 workers.add(w); // 獲取集合大小 int s = workers.size(); if (s > largestPoolSize)// 隊(duì)列大小大于largestPoolSize // 重新設(shè)置largestPoolSize largestPoolSize = s; // 設(shè)置worker已被添加標(biāo)志 workerAdded = true; } } finally { // 釋放鎖 mainLock.unlock(); } if (workerAdded) {// worker被添加 // 開始執(zhí)行worker的run方法 t.start(); // 設(shè)置worker已開始標(biāo)志 workerStarted = true; } } } finally { if (! workerStarted)// worker沒有開始 // 添加worker失敗 addWorkerFailed(w); } return workerStarted; }任務(wù)執(zhí)行過程
1、runworker方法
runWorker函數(shù)中會(huì)實(shí)際執(zhí)行給定任務(wù)(即調(diào)用用戶重寫的run方法),并且當(dāng)給定任務(wù)完成后,會(huì)繼續(xù)從阻塞隊(duì)列中取任務(wù),直到阻塞隊(duì)列為空(即任務(wù)全部完成).在執(zhí)行給定任務(wù)時(shí)會(huì)調(diào)用鉤子函數(shù)利用鉤子函數(shù)可以完成用戶自定義的一些邏輯,在runWorker中會(huì)調(diào)用getTask函數(shù)和processWorkerExit鉤子函數(shù)
final void runWorker(Worker w) { // 獲取當(dāng)前線程 Thread wt = Thread.currentThread(); // 獲取w的firstTask Runnable task = w.firstTask; w.firstTask = null; // 釋放鎖(設(shè)置state為0,允許中斷) w.unlock(); // allow interrupts boolean completedAbruptly = true; try { 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) ||// 線程池運(yùn)行狀態(tài)至少應(yīng)該高于STOP (Thread.interrupted() &&// 線程被中斷 runStateAtLeast(ctl.get(), STOP))) &&// 再次檢查,線程池的運(yùn)行狀態(tài)至少應(yīng)該高于STOP !wt.isInterrupted())// wt線程(當(dāng)前線程)沒有被中斷 wt.interrupt();// 中斷wt線程(當(dāng)前線程) try { // 在執(zhí)行之前調(diào)用鉤子函數(shù) beforeExecute(wt, task); Throwable thrown = null; try { // 運(yùn)行給定的任務(wù) 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 { // 執(zhí)行完后調(diào)用鉤子函數(shù) afterExecute(task, thrown); } } finally { task = null; // 增加給worker完成的任務(wù)數(shù)量 w.completedTasks++; // 釋放鎖 w.unlock(); } } completedAbruptly = false; } finally { // 處理完成后,調(diào)用鉤子函數(shù) processWorkerExit(w, completedAbruptly); } }
2.getTask方法
getTask函數(shù)用于從workerQueue阻塞隊(duì)列中獲取Runnable對(duì)象,由于是阻塞隊(duì)列,所以支持有限時(shí)間等待poll和無限時(shí)間等待take.在該函數(shù)中還會(huì)相應(yīng)shutdown和shutDownNow函數(shù)的操作,若檢測(cè)到線程池處于SHUTDOWN或STOP狀態(tài),則會(huì)返回null,而不再返回阻塞隊(duì)列中的Runnable對(duì)象
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) {// 無限循環(huán),確保操作成功 // 獲取線程池控制狀態(tài) int c = ctl.get(); // 運(yùn)行狀態(tài) int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {// 大于等于SHUTDOWN(表示調(diào)用了shutDown)并且-->大于等于STOP(調(diào)用shutDownNow或者worker阻塞隊(duì)列為空) // 減少worker數(shù)量 decrementWorkerCount(); // 返回null,不執(zhí)行任務(wù) return null; } // 獲取worker數(shù)量 int wc = workerCountOf(c); // Are workers subject to culling? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;// 是否允許coreThread超時(shí)或workerCount大于核心大小 if ((wc > maximumPoolSize || (timed && timedOut))// worker數(shù)量大于maxinumPoolSize && (wc > 1 || workQueue.isEmpty())) {// workerCount大于1或worker阻塞隊(duì)列為空(在阻塞隊(duì)列不為空時(shí),需要保證至少有一個(gè)wc) if (compareAndDecrementWorkerCount(c))// 比較并減少workerCount // 返回null,不執(zhí)行任務(wù),該worker會(huì)退出 return null; // 跳過剩余部分,繼續(xù)循環(huán) continue; } try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :// 等待指定時(shí)間 workQueue.take();// 一直等待,直到有元素 if (r != null) return r; // 等待指定時(shí)間后沒有獲取元素則超時(shí) timedOut = true; } catch (InterruptedException retry) { // 拋出了被中斷異常,重試沒有超時(shí) timedOut = false; } } }
3.processWorkerExit方法
processWorkerExit函數(shù)是在worker退出時(shí)調(diào)用到的鉤子函數(shù),而引起worker退出的主要因素如下:
1.阻塞隊(duì)列已經(jīng)為空,即沒有任務(wù)可以運(yùn)行了
2.調(diào)用了shutDown或shutDownNow函數(shù)
此函數(shù)會(huì)根據(jù)是否中斷了空閑線程來確定是否減少workerCount的值,并且將worker從workers集合中移除并且會(huì)嘗試終止線程池
private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) // 如果被中斷,則需要減少workCount decrementWorkerCount(); // 獲取可重入鎖 final ReentrantLock mainLock = this.mainLock; // 獲取鎖 mainLock.lock(); try { // 將worker完成的任務(wù)添加到總的完成任務(wù)中 completedTaskCount += w.completedTasks; // 從workers集合中移除該worker workers.remove(w); } finally { // 釋放鎖 mainLock.unlock(); } // 嘗試終止 tryTerminate(); // 獲取線程池控制狀態(tài) int c = ctl.get(); if (runStateLessThan(c, STOP)) {// 小于STOP的運(yùn)行狀態(tài) if (!completedAbruptly) { int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty())// 允許核心超時(shí)并且workQueue阻塞隊(duì)列不為空 min = 1; if (workerCountOf(c) >= min)// workerCount大于等于min return; // replacement not needed } // 添加worker addWorker(null, false); } }任務(wù)關(guān)閉過程
1.shutdown方法
shutdown會(huì)按過去執(zhí)行已提交任務(wù)的順序發(fā)起一個(gè)有序的關(guān)閉,但是不接受新任務(wù).首先檢查是否具有shutdown的權(quán)限,然后設(shè)置線程池的控制為SHUTDOWN,之后中斷空閑的worker,最后嘗試終止線程池.
public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 檢查shutdown權(quán)限 checkShutdownAccess(); // 設(shè)置線程控制狀態(tài)為SHUTDOWN advanceRunState(SHUTDOWN); // 中斷空閑worker interruptIdleWorkers(); // 調(diào)用shutdown鉤子函數(shù) onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } // 嘗試終止 tryTerminate(); }
2.tryTerminate方法
final void tryTerminate() { for (;;) {// 無限循環(huán),確保操作成功 // 獲取線程池控制狀態(tài) int c = ctl.get(); if (isRunning(c) ||// 線程池的運(yùn)行狀態(tài)為RUNNING runStateAtLeast(c, TIDYING) ||// 線程池的運(yùn)行狀態(tài)最大要大于TIDYING (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))// 線程池運(yùn)行狀態(tài)為SHUTDOWN并且workQueue隊(duì)列不為null // 不能終止,直接返回 return; if (workerCountOf(c) != 0) { // 線程池正在運(yùn)行的worker數(shù)量不為0 // 僅僅中斷一個(gè)空閑的worker interruptIdleWorkers(ONLY_ONE); return; } // 獲取線程池的鎖 final ReentrantLock mainLock = this.mainLock; // 獲取鎖 mainLock.lock(); try { if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {// 比較并設(shè)置線程池控制狀態(tài)為TIDYING try { // 終止,鉤子函數(shù) terminated(); } finally { ctl.set(ctlOf(TERMINATED, 0)); termination.signalAll(); } return; } } finally { // 釋放鎖 mainLock.unlock(); } // else retry on failed CAS } }
3.interruptIdleWorkers方法
private void interruptIdleWorkers(boolean onlyOne) { // 線程池的鎖 final ReentrantLock mainLock = this.mainLock; // 獲取鎖 mainLock.lock(); try { for (Worker w : workers) {// 遍歷workers隊(duì)列 // worker對(duì)應(yīng)的線程 Thread t = w.thread; if (!t.isInterrupted() && w.tryLock()) {// 線程未被中斷并且成功獲得鎖 try { // 中斷線程 t.interrupt(); } catch (SecurityException ignore) { } finally { // 釋放鎖 w.unlock(); } } if (onlyOne) break; } } finally { mainLock.unlock(); } }
文章版權(quán)歸作者所有,未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為,您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請(qǐng)注明本文地址:http://specialneedsforspecialkids.com/yun/76812.html
摘要:中的線程池是運(yùn)用場(chǎng)景最多的并發(fā)框架。才是真正的線程池。存放任務(wù)的隊(duì)列存放需要被線程池執(zhí)行的線程隊(duì)列。所以線程池的所有任務(wù)完成后,它最終會(huì)收縮到的大小。飽和策略一般情況下,線程池采用的是,表示無法處理新任務(wù)時(shí)拋出異常。 Java線程池 1. 簡(jiǎn)介 系統(tǒng)啟動(dòng)一個(gè)新線程的成本是比較高的,因?yàn)樗婕芭c操作系統(tǒng)的交互,這個(gè)時(shí)候使用線程池可以提升性能,尤其是需要?jiǎng)?chuàng)建大量聲明周期很短暫的線程時(shí)。Ja...
摘要:當(dāng)活動(dòng)線程核心線程非核心線程達(dá)到這個(gè)數(shù)值后,后續(xù)任務(wù)將會(huì)根據(jù)來進(jìn)行拒絕策略處理。線程池工作原則當(dāng)線程池中線程數(shù)量小于則創(chuàng)建線程,并處理請(qǐng)求。當(dāng)線程池中的數(shù)量等于最大線程數(shù)時(shí)默默丟棄不能執(zhí)行的新加任務(wù),不報(bào)任何異常。 spring-cache使用記錄 spring-cache的使用記錄,坑點(diǎn)記錄以及采用的解決方案 深入分析 java 線程池的實(shí)現(xiàn)原理 在這篇文章中,作者有條不紊的將 ja...
摘要:四種線程池的使用介紹的弊端及四種線程池的使用,線程池的作用線程池作用就是限制系統(tǒng)中執(zhí)行線程的數(shù)量。相比,提供的四種線程池的好處在于重用存在的線程,減少對(duì)象創(chuàng)建消亡的開銷,性能佳。延遲執(zhí)行描述創(chuàng)建一個(gè)定長(zhǎng)線程池,支持定時(shí)及周期性任務(wù)執(zhí)行。 java 四種線程池的使用 介紹new Thread的弊端及Java四種線程池的使用 1,線程池的作用 線程池作用就是限制系統(tǒng)中執(zhí)行線程的數(shù)量。 ...
系統(tǒng)啟動(dòng)一個(gè)線程的成本是比較高,使用線程池可以很好地提高性能,尤其是當(dāng)程序中需要?jiǎng)?chuàng)建大量生存期很短暫的線程時(shí) 線程池在系統(tǒng)啟動(dòng)時(shí)即創(chuàng)建大量空閑線程,將一個(gè)Runnable、Callable對(duì)象—–>傳給線程池—–>線程池啟動(dòng)里面的一個(gè)線程來執(zhí)行它們的run()或者call()方法———->當(dāng)線程執(zhí)行體執(zhí)行完成后,線程并不會(huì)死亡,而是再次返回線程池成為空閑狀態(tài),等待下一個(gè)Runnable、Calla...
摘要:高并發(fā)系列第篇文章。簡(jiǎn)單的說,在使用了線程池之后,創(chuàng)建線程變成了從線程池中獲取一個(gè)空閑的線程,然后使用,關(guān)閉線程變成了將線程歸還到線程池。如果調(diào)用了線程池的方法,線程池會(huì)提前把核心線程都創(chuàng)造好,并啟動(dòng)線程池允許創(chuàng)建的最大線程數(shù)。 java高并發(fā)系列第18篇文章。 本文主要內(nèi)容 什么是線程池 線程池實(shí)現(xiàn)原理 線程池中常見的各種隊(duì)列 自定義線程創(chuàng)建的工廠 常見的飽和策略 自定義飽和策略 ...
摘要:中的線程池運(yùn)用場(chǎng)景非常廣泛,幾乎所有的一步或者并發(fā)執(zhí)行程序都可以使用。代碼中如果執(zhí)行了方法,線程池會(huì)提前創(chuàng)建并啟動(dòng)所有核心線程。線程池最大數(shù)量線程池允許創(chuàng)建的線程最大數(shù)量。被稱為是可重用固定線程數(shù)的線程池。 Java中的線程池運(yùn)用場(chǎng)景非常廣泛,幾乎所有的一步或者并發(fā)執(zhí)行程序都可以使用。那么線程池有什么好處呢,以及他的實(shí)現(xiàn)原理是怎么樣的呢? 使用線程池的好處 在開發(fā)過程中,合理的使用線程...
閱讀 829·2021-11-22 11:59
閱讀 3245·2021-11-17 09:33
閱讀 2312·2021-09-29 09:34
閱讀 1944·2021-09-22 15:25
閱讀 1960·2019-08-30 15:55
閱讀 1325·2019-08-30 15:55
閱讀 536·2019-08-30 15:53
閱讀 3351·2019-08-29 13:55