java 中ThreadPoolExecutor原理分析
java 中ThreadPoolExecutor原理分析
線程池簡介
Java線程池是開發(fā)中常用的工具,當(dāng)我們有異步、并行的任務(wù)要處理時(shí),經(jīng)常會(huì)用到線程池,或者在實(shí)現(xiàn)一個(gè)服務(wù)器時(shí),也需要使用線程池來接收連接處理請(qǐng)求。
線程池使用
JDK中提供的線程池實(shí)現(xiàn)位于java.util.concurrent.ThreadPoolExecutor。在使用時(shí),通常使用ExecutorService接口,它提供了submit,invokeAll,shutdown等通用的方法。
在線程池配置方面,Executors類中提供了一些靜態(tài)方法能夠提供一些常用場(chǎng)景的線程池,如newFixedThreadPool,newCachedThreadPool,newSingleThreadExecutor等,這些方法最終都是調(diào)用到了ThreadPoolExecutor的構(gòu)造函數(shù)。
ThreadPoolExecutor的包含所有參數(shù)的構(gòu)造函數(shù)是
/** * @param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {@code allowCoreThreadTimeOut} is set * @param maximumPoolSize the maximum number of threads to allow in the * pool * @param keepAliveTime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating. * @param unit the time unit for the {@code keepAliveTime} argument * @param workQueue the queue to use for holding tasks before they are * executed. This queue will hold only the {@code Runnable} * tasks submitted by the {@code execute} method. * @param threadFactory the factory to use when the executor * creates a new thread * @param handler the handler to use when execution is blocked * because the thread bounds and queue capacities are reached public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, 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設(shè)置線程池的核心線程數(shù),當(dāng)添加新任務(wù)時(shí),如果線程池中的線程數(shù)小于corePoolSize,則不管當(dāng)前是否有線程閑置,都會(huì)創(chuàng)建一個(gè)新的線程來執(zhí)行任務(wù)。
- maximunPoolSize是線程池中允許的最大的線程數(shù)
- workQueue用于存放排隊(duì)的任務(wù)
- keepAliveTime是大于corePoolSize的線程閑置的超時(shí)時(shí)間
- handler用于在任務(wù)逸出、線程池關(guān)閉時(shí)的任務(wù)處理 ,線程池的線程增長策略為,當(dāng)前線程數(shù)小于corePoolSize時(shí),新增線程,當(dāng)線程數(shù)=corePoolSize且corePoolSize時(shí),只有在workQueue不能存放新的任務(wù)時(shí)創(chuàng)建新線程,超出的線程在閑置keepAliveTime后銷毀。
實(shí)現(xiàn)(基于JDK1.8)
ThreadPoolExecutor中保存的狀態(tài)有
當(dāng)前線程池狀態(tài), 包括RUNNING,SHUTDOWN,STOP,TIDYING,TERMINATED。
當(dāng)前有效的運(yùn)行線程的數(shù)量。
將這兩個(gè)狀態(tài)放到一個(gè)int變量中,前三位作為線程池狀態(tài),后29位作為線程數(shù)量。
例如0b11100000000000000000000000000001, 表示RUNNING, 一個(gè)線程。
通過HashSet來存儲(chǔ)工作者集合,訪問該HashSet前必須先獲取保護(hù)狀態(tài)的mainLock:ReentrantLock
submit、execute
execute的執(zhí)行方式為,首先檢查當(dāng)前worker數(shù)量,如果小于corePoolSize,則嘗試add一個(gè)core Worker。線程池在維護(hù)線程數(shù)量以及狀態(tài)檢查上做了大量檢測(cè)。
public void execute(Runnable command) { int c = ctl.get(); // 如果當(dāng)期數(shù)量小于corePoolSize if (workerCountOf(c) < corePoolSize) { // 嘗試增加worker if (addWorker(command, true)) return; c = ctl.get(); } // 如果線程池正在運(yùn)行并且成功添加到工作隊(duì)列中 if (isRunning(c) && workQueue.offer(command)) { // 再次檢查狀態(tài),如果已經(jīng)關(guān)閉則執(zhí)行拒絕處理 int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); // 如果工作線程都down了 else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); }
addWorker方法實(shí)現(xiàn)
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 || 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 } } 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) { // 如果添加成功,則啟動(dòng)該線程,執(zhí)行Worker的run方法,Worker的run方法執(zhí)行外部的runWorker(Worker) t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
Worker類繼承了AbstractQueuedSynchronizer獲得了同步等待這樣的功能。
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) { } } }
runWorker(Worker)是Worker的輪詢執(zhí)行邏輯,不斷地從工作隊(duì)列中獲取任務(wù)并執(zhí)行它們。Worker每次執(zhí)行任務(wù)前需要進(jìn)行l(wèi)ock,防止在執(zhí)行任務(wù)時(shí)被interrupt。
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; 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) || (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; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }
ThreadPoolExecutor的submit方法中將Callable包裝成FutureTask后交給execute方法。
FutureTask
FutureTask繼承于Runnable和Future,F(xiàn)utureTask定義的幾個(gè)狀態(tài)為
NEW, 尚未執(zhí)行
COMPLETING, 正在執(zhí)行
NORMAL, 正常執(zhí)行完成得到結(jié)果
EXCEPTIONAL, 執(zhí)行拋出異常
CANCELLED, 執(zhí)行被取消
INTERRUPTING,執(zhí)行正在被中斷
INTERRUPTED, 已經(jīng)中斷。
其中關(guān)鍵的get方法
public V get() throws InterruptedException, ExecutionException { int s = state; if (s <= COMPLETING) s = awaitDone(false, 0L); return report(s); }
先獲取當(dāng)前狀態(tài),如果還未執(zhí)行完成并且正常,則進(jìn)入等待結(jié)果流程。在awaitDone不斷循環(huán)獲取當(dāng)前狀態(tài),如果沒有結(jié)果,則將自己通過CAS的方式添加到等待鏈表的頭部,如果設(shè)置了超時(shí),則LockSupport.parkNanos到指定的時(shí)間。
static final class WaitNode { volatile Thread thread; volatile WaitNode next; WaitNode() { thread = Thread.currentThread(); } } private int awaitDone(boolean timed, long nanos) throws InterruptedException { final long deadline = timed ? System.nanoTime() + nanos : 0L; WaitNode q = null; boolean queued = false; for (;;) { if (Thread.interrupted()) { removeWaiter(q); throw new InterruptedException(); } int s = state; if (s > COMPLETING) { if (q != null) q.thread = null; return s; } else if (s == COMPLETING) // cannot time out yet Thread.yield(); else if (q == null) q = new WaitNode(); else if (!queued) queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q); else if (timed) { nanos = deadline - System.nanoTime(); if (nanos <= 0L) { removeWaiter(q); return state; } LockSupport.parkNanos(this, nanos); } else LockSupport.park(this); } }
FutureTask的run方法是執(zhí)行任務(wù)并設(shè)置結(jié)果的位置,首先判斷當(dāng)前狀態(tài)是否為NEW并且將當(dāng)前線程設(shè)置為執(zhí)行線程,然后調(diào)用Callable的call獲取結(jié)果后設(shè)置結(jié)果修改FutureTask狀態(tài)。
public void run() { if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())) return; try { Callable<V> c = callable; if (c != null && state == NEW) { V result; boolean ran; try { result = c.call(); ran = true; } catch (Throwable ex) { result = null; ran = false; setException(ex); } if (ran) set(result); } } finally { // runner must be non-null until state is settled to // prevent concurrent calls to run() runner = null; // state must be re-read after nulling runner to prevent // leaked interrupts int s = state; if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } }
感謝閱讀,希望能幫助到大家,謝謝大家對(duì)本站的支持!
相關(guān)文章
java文件操作工具類實(shí)現(xiàn)復(fù)制文件和文件合并
這篇文章主要介紹了java文件操作工具類,類實(shí)現(xiàn)了復(fù)制文件和文件合并的功能,需要的朋友可以參考下2014-03-03解讀Spring事務(wù)是如何實(shí)現(xiàn)的
這篇文章主要介紹了Spring事務(wù)是如何實(shí)現(xiàn)的問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-03-03SpringBoot項(xiàng)目鑒權(quán)的4種方式小結(jié)
本文主要介紹了SpringBoot項(xiàng)目鑒權(quán)的4種方式小結(jié),包括傳統(tǒng)AOP、攔截器、參數(shù)解析器和過濾器,文中通過示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2021-12-12SpringCloud Gateway網(wǎng)關(guān)功能介紹與使用
SpringCloud Gateway 是 Spring Cloud 的一個(gè)全新項(xiàng)目,它旨在為微服務(wù)架構(gòu)提供一種簡單有效的統(tǒng)一的 API 路由管理方式。這篇文章主要介紹了SpringCloud Gateway網(wǎng)關(guān)作用,需要的朋友可以參考下2022-12-12IDEA巧用Postfix Completion讓碼速起飛(小技巧)
這篇文章主要介紹了IDEA巧用Postfix Completion讓碼速起飛,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-08-08JavaSE基礎(chǔ)之反射機(jī)制(反射Class)詳解
反射機(jī)制有什么用?通過java語言中的反射機(jī)制可以操作字節(jié)碼文件,可以讀和修改字節(jié)碼文件。所以本文將為大家講講反射機(jī)制的使用,需要的可以參考一下2022-09-09SpringBoot+Vue+Element-ui實(shí)現(xiàn)前后端分離
使用前后端分離的方式,可以減少代碼耦合,本文主要介紹了SpringBoot+Vue+Element-ui實(shí)現(xiàn)前后端分離,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-06-06