今天就跟大家聊聊有關(guān)Java中線程池的執(zhí)行原理是什么,可能很多人都不太了解,為了讓大家更加了解,小編給大家總結(jié)了以下內(nèi)容,希望大家根據(jù)這篇文章可以有所收獲。
成都創(chuàng)新互聯(lián)專注于企業(yè)網(wǎng)絡(luò)營銷推廣、網(wǎng)站重做改版、都昌網(wǎng)站定制設(shè)計(jì)、自適應(yīng)品牌網(wǎng)站建設(shè)、html5、商城建設(shè)、集團(tuán)公司官網(wǎng)建設(shè)、外貿(mào)營銷網(wǎng)站建設(shè)、高端網(wǎng)站制作、響應(yīng)式網(wǎng)頁設(shè)計(jì)等建站業(yè)務(wù),價(jià)格優(yōu)惠性價(jià)比高,為都昌等各大城市提供網(wǎng)站開發(fā)制作服務(wù)。
線程池狀態(tài)
首先認(rèn)識(shí)兩個(gè)貫穿線程池代碼的參數(shù):
runState:線程池運(yùn)行狀態(tài)
workerCount:工作線程的數(shù)量
線程池用一個(gè)32位的int來同時(shí)保存runState和workerCount,其中高3位是runState,其余29位是workerCount。代碼中會(huì)反復(fù)使用runStateOf和workerCountOf來獲取runState和workerCount。
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; // 線程池狀態(tài) 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; // ctl操作 private static int runStateOf(int c) { return c & ~CAPACITY; } private static int workerCountOf(int c) { return c & CAPACITY; } private static int ctlOf(int rs, int wc) { return rs | wc; }
RUNNING:可接收新任務(wù),可執(zhí)行等待隊(duì)列里的任務(wù)
SHUTDOWN:不可接收新任務(wù),可執(zhí)行等待隊(duì)列里的任務(wù)
STOP:不可接收新任務(wù),不可執(zhí)行等待隊(duì)列里的任務(wù),并且嘗試終止所有在運(yùn)行任務(wù)
TIDYING:所有任務(wù)已經(jīng)終止,執(zhí)行terminated()
TERMINATED:terminated()執(zhí)行完成
線程池狀態(tài)默認(rèn)從RUNNING開始流轉(zhuǎn),到狀態(tài)TERMINATED結(jié)束,中間不需要經(jīng)過每一種狀態(tài),但不能讓狀態(tài)回退。下面是狀態(tài)變化可能的路徑和變化條件:
Worker的創(chuàng)建
線程池是由Worker類負(fù)責(zé)執(zhí)行任務(wù),Worker繼承了AbstractQueuedSynchronizer,引出了Java并發(fā)框架的核心AQS。
AbstractQueuedSynchronizer,簡(jiǎn)稱AQS,是Java并發(fā)包里一系列同步工具的基礎(chǔ)實(shí)現(xiàn),原理是根據(jù)狀態(tài)位來控制線程的入隊(duì)阻塞、出隊(duì)喚醒來處理同步。
AQS不會(huì)在這里展開討論,只需要知道Worker包裝了Thread,由它去執(zhí)行任務(wù)。
調(diào)用execute將會(huì)根據(jù)線程池的情況創(chuàng)建Worker,可以歸納出下圖四種情況:
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); //1 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } //2 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) //3 reject(command); else if (workerCountOf(recheck) == 0) //4 addWorker(null, false); } //5 else if (!addWorker(command, false)) //6 reject(command); }
標(biāo)記1對(duì)應(yīng)第一種情況,要留意addWorker傳入了core,core=true為corePoolSize,core=false為maximumPoolSize,
新增時(shí)需要檢查workerCount是否超過允許的最大值。
標(biāo)記2對(duì)應(yīng)第二種情況,檢查線程池是否在運(yùn)行,并且將任務(wù)加入等待隊(duì)列。標(biāo)記3再檢查一次線程池狀態(tài),如果線程池忽然處于非運(yùn)行狀態(tài),那就將等待隊(duì)列剛加的任務(wù)刪掉,再交給RejectedExecutionHandler處理。標(biāo)記4發(fā)現(xiàn)沒有worker,就先補(bǔ)充一個(gè)空任務(wù)的worker。
標(biāo)記5對(duì)應(yīng)第三種情況,等待隊(duì)列不能再添加任務(wù)了,調(diào)用addWorker添加一個(gè)去處理。
標(biāo)記6對(duì)應(yīng)第四種情況,addWorker的core傳入false,返回調(diào)用失敗,代表workerCount已經(jīng)超出maximumPoolSize,那就交給RejectedExecutionHandler處理。
private boolean addWorker(Runnable firstTask, boolean core) { //1 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 || 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 } } //2 boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { 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) { t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
標(biāo)記1的第一段代碼,目的很簡(jiǎn)單,是為workerCount加一。至于為什么代碼寫了這么長(zhǎng),是因?yàn)榫€程池的狀態(tài)在不斷
變化,并發(fā)環(huán)境下需要保證變量的同步性。外循環(huán)判斷線程池狀態(tài)、任務(wù)非空和隊(duì)列非空,內(nèi)循環(huán)使用CAS機(jī)制保證workerCount正確地遞增。不了解CAS可以看認(rèn)識(shí)非阻塞的同步機(jī)制CAS,后續(xù)增減workerCount都會(huì)使用CAS。
標(biāo)記2的第二段代碼,就比較簡(jiǎn)單。創(chuàng)建一個(gè)新Worker對(duì)象,將Worker添加進(jìn)workers里(Set集合)。成功添加后,啟動(dòng)worker里的線程。在finally里判斷線程是否啟動(dòng)成功,不成功直接調(diào)用addWorkerFailed。
private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (w != null) workers.remove(w); decrementWorkerCount(); tryTerminate(); } finally { mainLock.unlock(); } }
addWorkerFailed將減少已經(jīng)遞增的workerCount,并且調(diào)用tryTerminate結(jié)束線程池。
Worker的執(zhí)行
Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } public void run() { runWorker(this); }
Worker在構(gòu)造函數(shù)里采用ThreadFactory創(chuàng)建Thread,在run方法里調(diào)用了runWorker,看來是真正執(zhí)行任務(wù)的地方。
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 while (task != null || (task = getTask()) != null) { w.lock(); //2 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { //3 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; //4 w.completedTasks++; w.unlock(); } } completedAbruptly = false; //5 } finally { //6 processWorkerExit(w, completedAbruptly); } }
標(biāo)記1進(jìn)入循環(huán),從getTask獲取要執(zhí)行的任務(wù),直到返回null。這里達(dá)到了線程復(fù)用的效果,讓線程處理多個(gè)任務(wù)。
標(biāo)記2是一個(gè)比較復(fù)雜的判斷,保證了線程池在STOP狀態(tài)下線程是中斷的,非STOP狀態(tài)下線程沒有被中斷。如果你不了解Java的中斷機(jī)制,看如何正確結(jié)束Java線程這篇。
標(biāo)記3調(diào)用了run方法,真正執(zhí)行了任務(wù)。執(zhí)行前后提供了beforeExecute和afterExecute兩個(gè)方法,由子類實(shí)現(xiàn)。
標(biāo)記4里的completedTasks統(tǒng)計(jì)worker執(zhí)行了多少任務(wù),最后累加進(jìn)completedTaskCount變量,可以調(diào)用相應(yīng)方法返回一些統(tǒng)計(jì)信息。
標(biāo)記5的變量completedAbruptly表示worker是否異常終止,執(zhí)行到這里代表執(zhí)行正常,后續(xù)的方法需要這個(gè)變量。
標(biāo)記6調(diào)用processWorkerExit結(jié)束,后面會(huì)分析。
接著來看worker從等待隊(duì)列獲取任務(wù)的getTask方法:
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get(); int rs = runStateOf(c); //1 // Check if queue empty only if necessary. if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); //2 // Are workers subject to culling? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } //3 try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
標(biāo)記1檢查線程池的狀態(tài),這里就體現(xiàn)出SHUTDOWN和STOP的區(qū)別。如果線程池是SHUTDOWN狀態(tài),還會(huì)先處理完等待隊(duì)列的任務(wù);如果是STOP狀態(tài),就不再處理等待隊(duì)列里的任務(wù)了。
標(biāo)記2先看allowCoreThreadTimeOut這個(gè)變量,false時(shí)worker空閑,也不會(huì)結(jié)束;true時(shí),如果worker空閑超過keepAliveTime,就會(huì)結(jié)束。接著是一個(gè)很復(fù)雜的判斷,好難轉(zhuǎn)成文字描述,自己看吧。注意一下wc>maximumPoolSize,出現(xiàn)這種可能是在運(yùn)行中調(diào)用setMaximumPoolSize,還有wc>1,在等待隊(duì)列非空時(shí),至少保留一個(gè)worker。
標(biāo)記3是從等待隊(duì)列取任務(wù)的邏輯,根據(jù)timed分為等待keepAliveTime或者阻塞直到有任務(wù)。
最后來看結(jié)束worker需要執(zhí)行的操作:
private void processWorkerExit(Worker w, boolean completedAbruptly) { //1 if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount(); //2 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { completedTaskCount += w.completedTasks; workers.remove(w); } finally { mainLock.unlock(); } //3 tryTerminate(); int c = ctl.get(); //4 if (runStateLessThan(c, STOP)) { if (!completedAbruptly) { int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) min = 1; if (workerCountOf(c) >= min) return; // replacement not needed } addWorker(null, false); } }
正常情況下,在getTask里就會(huì)將workerCount減一。標(biāo)記1處用變量completedAbruptly判斷worker是否異常退出,如果是,需要補(bǔ)充對(duì)workerCount的減一。
標(biāo)記2將worker處理任務(wù)的數(shù)量累加到總數(shù),并且在集合workers中去除。
標(biāo)記3嘗試終止線程池,后續(xù)會(huì)研究。
標(biāo)記4處理線程池還是RUNNING或SHUTDOWN狀態(tài)時(shí),如果worker是異常結(jié)束,那么會(huì)直接addWorker。如果allowCoreThreadTimeOut=true,并且等待隊(duì)列有任務(wù),至少保留一個(gè)worker;如果allowCoreThreadTimeOut=false,workerCount不少于corePoolSize。
總結(jié)一下worker:線程池啟動(dòng)后,worker在池內(nèi)創(chuàng)建,包裝了提交的Runnable任務(wù)并執(zhí)行,執(zhí)行完就等待下一個(gè)任務(wù),不再需要時(shí)就結(jié)束。
線程池的關(guān)閉
線程池的關(guān)閉不是一關(guān)了事,worker在池里處于不同狀態(tài),必須安排好worker的”后事”,才能真正釋放線程池。ThreadPoolExecutor提供兩種方法關(guān)閉線程池:
shutdown:不能再提交任務(wù),已經(jīng)提交的任務(wù)可繼續(xù)運(yùn)行;
shutdownNow:不能再提交任務(wù),已經(jīng)提交但未執(zhí)行的任務(wù)不能運(yùn)行,在運(yùn)行的任務(wù)可繼續(xù)運(yùn)行,但會(huì)被中斷,返回已經(jīng)提交但未執(zhí)行的任務(wù)。
public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); //1 安全策略機(jī)制 advanceRunState(SHUTDOWN); //2 interruptIdleWorkers(); //3 onShutdown(); //4 空方法,子類實(shí)現(xiàn) } finally { mainLock.unlock(); } tryTerminate(); //5 }
shutdown將線程池切換到SHUTDOWN狀態(tài),并調(diào)用interruptIdleWorkers請(qǐng)求中斷所有空閑的worker,最后調(diào)用tryTerminate嘗試結(jié)束線程池。
public List<Runnable> shutdownNow() { List<Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(STOP); interruptWorkers(); tasks = drainQueue(); //1 } finally { mainLock.unlock(); } tryTerminate(); return tasks; }
shutdownNow和shutdown類似,將線程池切換為STOP狀態(tài),中斷目標(biāo)是所有worker。drainQueue會(huì)將等待隊(duì)列里未執(zhí)行的任務(wù)返回。
interruptIdleWorkers和interruptWorkers實(shí)現(xiàn)原理都是遍歷workers集合,中斷條件符合的worker。
上面的代碼多次出現(xiàn)調(diào)用tryTerminate,這是一個(gè)嘗試將線程池切換到TERMINATED狀態(tài)的方法。
final void tryTerminate() { for (;;) { int c = ctl.get(); //1 if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return; //2 if (workerCountOf(c) != 0) { // Eligible to terminate interruptIdleWorkers(ONLY_ONE); return; } //3 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { try { terminated(); } finally { ctl.set(ctlOf(TERMINATED, 0)); termination.signalAll(); } return; } } finally { mainLock.unlock(); } // else retry on failed CAS } }
標(biāo)記1檢查線程池狀態(tài),下面幾種情況,后續(xù)操作都沒有必要,直接return。
RUNNING(還在運(yùn)行,不能停)
TIDYING或TERMINATED(已經(jīng)沒有在運(yùn)行的worker)
SHUTDOWN并且等待隊(duì)列非空(執(zhí)行完才能停)
標(biāo)記2在worker非空的情況下又調(diào)用了interruptIdleWorkers,你可能疑惑在shutdown時(shí)已經(jīng)調(diào)用過了,為什么又調(diào)用,而且每次只中斷一個(gè)空閑worker?
你需要知道,shutdown時(shí)worker可能在執(zhí)行中,執(zhí)行完阻塞在隊(duì)列的take,不知道要結(jié)束,所有要補(bǔ)充調(diào)用interruptIdleWorkers。每次只中斷一個(gè)是因?yàn)閜rocessWorkerExit時(shí),還會(huì)執(zhí)行tryTerminate,自動(dòng)中斷下一個(gè)空閑的worker。
標(biāo)記3是最終的狀態(tài)切換。線程池會(huì)先進(jìn)入TIDYING狀態(tài),再進(jìn)入TERMINATED狀態(tài),中間提供了terminated這個(gè)空方法供子類實(shí)現(xiàn)。
調(diào)用關(guān)閉線程池方法后,需要等待線程池切換到TERMINATED狀態(tài)。awaitTermination檢查限定時(shí)間內(nèi)線程池是否進(jìn)入TERMINATED狀態(tài),代碼如下:
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { long nanos = unit.toNanos(timeout); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (;;) { if (runStateAtLeast(ctl.get(), TERMINATED)) return true; if (nanos <= 0) return false; nanos = termination.awaitNanos(nanos); } } finally { mainLock.unlock(); } }
看完上述內(nèi)容,你們對(duì)Java中線程池的執(zhí)行原理是什么有進(jìn)一步的了解嗎?如果還想了解更多知識(shí)或者相關(guān)內(nèi)容,請(qǐng)關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道,感謝大家的支持。
網(wǎng)站名稱:Java中線程池的執(zhí)行原理是什么
文章位置:http://www.rwnh.cn/article34/igpcse.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供網(wǎng)站設(shè)計(jì)公司、自適應(yīng)網(wǎng)站、網(wǎng)站建設(shè)、做網(wǎng)站、域名注冊(cè)、定制開發(fā)
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請(qǐng)盡快告知,我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如需處理請(qǐng)聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來源: 創(chuàng)新互聯(lián)