摘要:線程池技術旨在解決兩個不同的問題在處理大量異步任務時可以提高性能,因為減少了線程的銷毀,新建,切換等消耗性能的操作。線程池還有能力統一管理,調度,監控,調優線程等,還提供了一下基本的統計,比如已完成的任務數量。線程數量,線程池的狀態。
了解ThreadPoolExecutor
先看一下線程池類的類圖關系:
Executor接口Executor作者描述的是Executor提供了一種解耦方式將任務的提交和任務以何種技術執行分離;
Executor接口只有一個方法:
void execute(Runnable command);
execute方法接收一個Runnable對象,方法的描述是在未來的某個時間執行command。不管是在一個新的線程中執行,還是在線程池中執行,甚至在調用者線程中立即執行。
ExecutorService接口ExecutorService繼承了Executor接口,ExecutorService可以被關閉,關閉以后不再接收新的任務。ExecutorService提供了兩個不同的方法關閉ExecutorService。shutdown方法會等待之前還未執行的任務執行完畢再關閉,而shutdownNow則不會再啟動新的任務,還會中斷正在執行的任務。一旦關閉后,ExecutorService就不會有正在執行的任務,也不會有等待被執行的任務,更不會有新的任務被提交。ExecutorService關閉后應該處理好一些資源的回收。
ThreadPoolExecutor線程池技術旨在解決兩個不同的問題:
在處理大量異步任務時可以提高性能,因為減少了線程的銷毀,新建,切換等消耗性能的操作。
線程池還有能力統一管理,調度,監控,調優線程等,還提供了一下基本的統計,比如已完成的任務數量。
重要的狀態和狀態判斷的方法private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); //高3位和低29位分別表示狀態和線程數 private static final int COUNT_BITS = Integer.SIZE - 3; //1左移29位減一得到低29位都是1,即線程的最大數量,大概5億多 private static final int CAPACITY = (1 << COUNT_BITS) - 1; // runState is stored in the high-order bits private static final int RUNNING = -1 << COUNT_BITS;//111 private static final int SHUTDOWN = 0 << COUNT_BITS;//000 private static final int STOP = 1 << COUNT_BITS;//001 private static final int TIDYING = 2 << COUNT_BITS;//010 private static final int TERMINATED = 3 << COUNT_BITS;//011 // Packing and unpacking ctl //獲得狀態 private static int runStateOf(int c) { return c & ~CAPACITY; } //獲得線程數量 private static int workerCountOf(int c) { return c & CAPACITY; } //通過狀態和線程數量組裝ctl 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. */ //c狀態是否小于s狀態 private static boolean runStateLessThan(int c, int s) { return c < s; } //c狀態是否大于等于s狀態 private static boolean runStateAtLeast(int c, int s) { return c >= s; } //線程池是否是運行狀態 private static boolean isRunning(int c) { return c < SHUTDOWN; }
整個類最重要的一個狀態標志ctl是一個AtomicInteger,它包含了兩個字段的含義。workerCount線程數量,runState線程池的狀態。
這一個字段是如何包含兩個字段的含義的呢,Doug Lea大牛使用了一個int的32位bits的高三位保存了狀態值,低29位保存了線程數量。
其中五個狀態:
RUNNING:接收新的任務,處理隊列中的任務;
SHUTDOWN:不接收新的任務,但處理隊列中的任務;
STOP:不接收新的任務,不處理隊列中的任務,中斷正在執行的任務;
TIDYING:所有任務都終止,線程數為0, 線程過度到TIDYING時會調用terminated鉤子方法;
TERMINATED:terminated執行完畢;
狀態之間的轉換:
RUNNING -> SHUTDOWN:調用shutdown方法;
(RUNNING or SHUTDOWN) -> STOP:調用shutdownNow方法;
SHUTDOWN -> TIDYING:當線程池和任務隊列都為空;
STOP -> TIDYING:當線程池為空;
TIDYING -> TERMINATED:當terminated方法執行完畢;
Worker類主要包含了線程運行任務時的終端控制狀態,同時還有一些少量的信息記錄。Worker適時的繼承了AQS,讓線程在任務執行之間獲取鎖和釋放鎖變得簡單。這確保了中斷是喚醒一個等待任務的線程,而不是中斷一個正在運行的任務線程。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */ private static final long serialVersionUID = 6138294804551838833L; /** Thread this worker is running in. Null if factory fails. */ final Thread thread; /** Initial task to run. Possibly null. */ Runnable firstTask; /** Per-thread task counter */ volatile long completedTasks; /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */ 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); } // Lock methods // // The value 0 represents the unlocked state. // The value 1 represents the locked state. protected boolean isHeldExclusively() { return getState() != 0; } protected boolean tryAcquire(int unused) { if (compareAndSetState(0, 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } protected boolean tryRelease(int unused) { setExclusiveOwnerThread(null); setState(0); return true; } public void lock() { acquire(1); } public boolean tryLock() { return tryAcquire(1); } public void unlock() { release(1); } public boolean isLocked() { return isHeldExclusively(); } void interruptIfStarted() { Thread t; if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try { t.interrupt(); } catch (SecurityException ignore) { } } } }
Worker繼承了AQS,實現了Runnable接口;在構造函數中,初始化了它的第一次仍無,使用threadFactory創建一個新的線程;
Worker繼承AQS,目的是想使用獨占鎖來表示線程是否正在執行任務,Worker的線程獲取了獨占鎖就說明它在執行任務,不能被中斷。從tryAcquire方法可以看出,它實現的是不可重入鎖,因為是否獲得鎖在這里表示一個狀態,如果可以重入的話,獨占鎖就失去了只表示一個狀態的含義。在構造函數初始化時,Worker將state設置為-1,因為在tryAcquire中CAS操作compareAndSetState(0, 1),表示state在-1時不能被中斷。在runWorker中將state設置為0.
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueworkQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }
說明一下各參數的含義:
corePoolSize:核心線程數量,即使線程是空閑的也保持在線程池中,除非allowCoreThreadTimeOut參數被設置;
maximumPoolSize:最大線程數量;
keepAliveTime:當線程數量超過核心線程數量時,超出的空閑線程等待新任務的最大時長;
unit:時間單位;
workQueue:存放將要被執行的任務的隊列;
threadFactory:創建線程的線程工廠;
handler:當任務隊列滿且沒有空閑的線程時處理任務的handler,線程池提供了四種策略:
AbortPolicy:直接拋出異常,默認;
CallerRunsPolicy:使用調用者的線程執行;
DiscardOldestPolicy:拋棄隊列最前的任務,執行當前任務;
DiscardPolicy:直接丟棄任務;
這些參數對整個線程池運行非常重要;
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. */ //獲取ctl int c = ctl.get(); //如果線程數小于核心線程數 if (workerCountOf(c) < corePoolSize) { //添加線程并執行任務 if (addWorker(command, true)) return; c = ctl.get(); } //線程數大于核心線程數 //如果線程池running狀態且添加任務到隊列成功 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); //如果線程池不是運行狀態,隊列移除任務,使用拒絕策略處理任務 if (! isRunning(recheck) && remove(command)) reject(command); //如果這時線程數為0,添加任務 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //隊列滿,添加線程失敗,使用拒絕策略處理任務 else if (!addWorker(command, false)) reject(command); }
在線程池添
數量如果小于核心線程數,則添加新的線程并執行當前任務,否則判斷如果隊列是否未滿,則添加當前任務到隊列,否則判斷線程數量如果小于最大線程數,則添加新的線程并執行,否則使用拒絕策略處理當前任務。
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. //如果運行狀態大于等于SHUTDOWN,不再接受新的任務,返回false //如果運行狀態等于SHUTDOWN且firstTask不為空,繼續執行下去,如果firstTask為空,queue為空,返回false,否則繼續執行;只要SHUTDOWN狀態下還有任務在,就需要往下執行,可能需要新建worker執行 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { //獲得線程數量 int wc = workerCountOf(c); //如果線程數量大于容量或者當core為true時wc大于等于核心線程數,當core為falsewc大于等于最大線程數量時,返回false if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //CAS線程數加一,成功則中斷循環 if (compareAndIncrementWorkerCount(c)) break retry; //如果CAS失敗,重新獲取ctl,線程池運行狀態沒變的話繼續loop 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 { //新建一個worker w = new Worker(firstTask); //能得到worker的thread 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()); //如果rs是RUNNING或者SHUTDOWN且firstTask為null //因為SHUTDOWN時還需要執行queue中的任務 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); //往線程池中添加worker workers.add(w); int s = workers.size(); //記錄線程池出現的最大線程數量 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { //啟動worker t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
worker的run方法調用的是runWorker;
runWorker方法final void runWorker(Worker w) { Thread wt = Thread.currentThread(); //保存worker的第一個任務 Runnable task = w.firstTask; //清空worker的第一個任務 w.firstTask = null; //這里將worker的state設置為0,允許中斷 w.unlock(); // allow interrupts boolean completedAbruptly = true; try { //如果task為空,則從隊列中獲取任務 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 //如果當前狀態大于等于STOP要保持當前線程中斷 //如果當前線程小于STOP即RUNNING或者SHUTDOWN,調用Thread.interrupted()清空中斷標志,如果這時調用了shutdownNow狀態為STOP,還是要保持中斷狀態 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { //執行任務前做的事 beforeExecute(wt, task); Throwable thrown = null; try { //執行任務 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; //worker的完成任務數量加一,此時是線程安全的 w.completedTasks++; //釋放鎖 w.unlock(); } } completedAbruptly = false; } finally { //線程退出 processWorkerExit(w, completedAbruptly); } }
每個task在調用runWorker后會一直循環執行任務,直到queue中沒有任務了,循環結束,worker生命周期結束。
getTask上面runWorker時調用了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. //如果rs大于等于SHUTDOWN,當RS大于等于STOP說明線程池已經不處理隊列中的任務了,當rs為SHUTDOWN時,如果隊列是空的,返回null if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { //線程數減一 decrementWorkerCount(); return null; } int wc = workerCountOf(c); // Are workers subject to culling? //是否超時控制,allowCoreThreadTimeOut默認false,代表不允許核心線程超時,對于超出核心線程的線程需要控制超時 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; //當線程數大于最大線程數,或者需要超時控制且上次獲取任務超時 //且線程數大于1或者隊列為空,嘗試將線程數減一并返回null if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; //失敗重試 continue; } try { //當需要超時控制時,在keepAliveTime時間內沒有獲取到任務的話返回null,否則調用take獲取任務,此時線程時阻塞的 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
getTask方法在線程數量大于核心線程數時會判斷在獲取task時進行超時判斷(poll),超時返回null這時getTask返回null,那當前worker的loop結束即run方法結束,線程生命周期結束。而核心線程則會調用take方法,當沒有任務時會阻塞。
processWorkerExitrunTask方法最后會調用processWorkerExit方法進行一些cleanup工作。
private void processWorkerExit(Worker w, boolean completedAbruptly) { //completedAbruptly為true時代表發生了異常,線程數減一 if (completedAbruptly) // If abrupt, then workerCount wasn"t adjusted decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //統計完成任務數 completedTaskCount += w.completedTasks; //線程池移除當前worker workers.remove(w); } finally { mainLock.unlock(); } // 根據線程池狀態進行判斷是否結束線程池 tryTerminate(); int c = ctl.get(); //當線程池狀態為RUNNING或者SHUTDOWN時 //如果發生異常,重新加入一個worker replacement if (runStateLessThan(c, STOP)) { if (!completedAbruptly) { //當allowCoreThreadTimeOut為true,最少要一個worker int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) min = 1; //當線程數大于等于最少需要的線程數,則不需要add新的worker if (workerCountOf(c) >= min) return; // replacement not needed } addWorker(null, false); } }tryTerminate方法
上面我們跳過了tryTerminate方法,該方法判斷是否要結束線程池,這里看一下
final void tryTerminate() { for (;;) { int c = ctl.get(); //當線程池狀態時RUNNING或者已經TIDYING或者已經TERMINATED或者SHUTDOWN且還有任務沒有被執行,直接返回 if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return; // 如果線程數不為0,則中斷一個空閑的工作線程 if (workerCountOf(c) != 0) { // Eligible to terminate //workQueue.take()時如果queue一直為空的話,線程會一直阻塞 interruptIdleWorkers(ONLY_ONE); return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //如果狀態設置成功為TIDYING,調用勾子方法terminated,該方法留給了子類實現 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { try { terminated(); } finally { //設置狀態為TERMINATED ctl.set(ctlOf(TERMINATED, 0)); termination.signalAll(); } return; } } finally { mainLock.unlock(); } // else retry on failed CAS } }interruptIdleWorkers
上面說為了當隊列一直為空的時候,核心線程會一直阻塞,所以調用了interruptIdleWorkers,我們看一下執行了什么:
private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) { Thread t = w.thread; if (!t.isInterrupted() && w.tryLock()) { try { t.interrupt(); } catch (SecurityException ignore) { } finally { w.unlock(); } } if (onlyOne) break; } } finally { mainLock.unlock(); } }
遍歷線程池中所有的線程,若線程沒有被中斷tryLock成功,就中斷該線程,LockSupport.park()能響應中斷信號,阻塞的線程被中斷喚醒。
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/70949.html
摘要:源碼分析創建可緩沖的線程池。源碼分析使用創建線程池源碼分析的構造函數構造函數參數核心線程數大小,當線程數,會創建線程執行最大線程數,當線程數的時候,會把放入中保持存活時間,當線程數大于的空閑線程能保持的最大時間。 之前創建線程的時候都是用的 newCachedThreadPoo,newFixedThreadPool,newScheduledThreadPool,newSingleThr...
摘要:當活動線程核心線程非核心線程達到這個數值后,后續任務將會根據來進行拒絕策略處理。線程池工作原則當線程池中線程數量小于則創建線程,并處理請求。當線程池中的數量等于最大線程數時默默丟棄不能執行的新加任務,不報任何異常。 spring-cache使用記錄 spring-cache的使用記錄,坑點記錄以及采用的解決方案 深入分析 java 線程池的實現原理 在這篇文章中,作者有條不紊的將 ja...
摘要:任務性質不同的任務可以用不同規模的線程池分開處理。線程池在運行過程中已完成的任務數量。如等于線程池的最大大小,則表示線程池曾經滿了。線程池的線程數量。獲取活動的線程數。通過擴展線程池進行監控。框架包括線程池,,,,,,等。 Java線程池 [toc] 什么是線程池 線程池就是有N個子線程共同在運行的線程組合。 舉個容易理解的例子:有個線程組合(即線程池,咱可以比喻為一個公司),里面有3...
摘要:線程池常見實現線程池一般包含三個主要部分調度器決定由哪個線程來執行任務執行任務所能夠的最大耗時等線程隊列存放并管理著一系列線程這些線程都處于阻塞狀態或休眠狀態任務隊列存放著用戶提交的需要被執行的任務一般任務的執行的即先提交的任務先被執行調度 線程池常見實現 線程池一般包含三個主要部分: 調度器: 決定由哪個線程來執行任務, 執行任務所能夠的最大耗時等 線程隊列: 存放并管理著一系列線...
摘要:當面試官問線程池時,你應該知道些什么一執行流程與不同,向中提交任務的時候,任務被包裝成對象加入延遲隊列并啟動一個線程。當我們創建出一個調度線程池以后,就可以開始提交任務了。 最近新接手的項目里大量使用了ScheduledThreadPoolExecutor類去執行一些定時任務,之前一直沒有機會研究這個類的源碼,這次趁著機會好好研讀一下。 原文地址:http://www.jianshu....
閱讀 3162·2023-04-25 17:19
閱讀 616·2021-11-23 09:51
閱讀 1339·2021-11-08 13:19
閱讀 776·2021-09-29 09:34
閱讀 1673·2021-09-28 09:36
閱讀 1494·2021-09-22 14:59
閱讀 2708·2019-08-29 16:38
閱讀 2053·2019-08-26 13:40