Java線程池execute()方法源碼全面解析

更新時間：2022年03月23日 12:10:36 作者：tydhot

這篇文章主要介紹了Java線程池execute()方法源碼全面解析，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

先看作者給出的注釋來理解線程池到底有什么作用

* Thread pools address two different problems: they usually
* provide improved performance when executing large numbers of
* asynchronous tasks, due to reduced per-task invocation overhead,
* and they provide a means of bounding and managing the resources,
* including threads, consumed when executing a collection of tasks.
* Each {@code ThreadPoolExecutor} also maintains some basic
* statistics, such as the number of completed tasks.

線程池處理了兩個不同的問題，線程池通過減少線程正式調(diào)用之前的開銷來給大量異步任務(wù)更優(yōu)秀的表現(xiàn)，與此同時給出了一系列綁定管理任務(wù)線程的一種手段。每個線程池都包含了一些基本信息，比如內(nèi)部完成的任務(wù)數(shù)量。

先看ThreadPoolExecutor類的一系列代表狀態(tài)的

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;
?
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;
?
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; }

ctl作為AtomicInteger類存放了類中的兩種信息，在其中由高3位來保存線程池的狀態(tài)，后29位來保存此時線程池中的Woker類線程數(shù)量(由此可知，線程池中的線程數(shù)量最高可以接受大約在五億左右)。由此可見給出的runStateOf()和workerCountOf()方法分別給出了查看線程狀態(tài)和線程數(shù)量的方法。

該類一共給出了五種狀態(tài)

讓我們看作者給出的注釋

* ? RUNNING: ?Accept new tasks and process queued tasks
* ? SHUTDOWN: Don't accept new tasks, but process queued tasks
* ? STOP: ? ? Don't accept new tasks, don't process queued tasks,
* ? ? ? ? ? ? and interrupt in-progress tasks
* ? TIDYING: ?All tasks have terminated, workerCount is zero,
* ? ? ? ? ? ? the thread transitioning to state TIDYING
* ? ? ? ? ? ? will run the terminated() hook method
* ? TERMINATED: terminated() has completed

RUNNING狀態(tài)可以接受新進(jìn)來的任務(wù)，同時也會執(zhí)行隊列里的任務(wù)。
SHUTDOWN 狀態(tài)已經(jīng)不會再接受新任務(wù)，但仍舊會處理隊列中的任務(wù)。
STOP狀態(tài)在之前的基礎(chǔ)上，不會處理隊列中的人物，在執(zhí)行的任務(wù)也會直接被打斷。
TIDYING狀態(tài)在之前的基礎(chǔ)上，所有任務(wù)都已經(jīng)終止，池中的Worker線程都已經(jīng)為0，也就是stop狀態(tài)在清理完所有工作線程之后就會進(jìn)入該狀態(tài)，同時在shutdown狀態(tài)在隊列空以及工作線程清理完畢之后也會直接進(jìn)入這個階段，這一階段會循環(huán)執(zhí)行terminated()方法。
TERMINATED 狀態(tài)作為最后的狀態(tài)，在之前的基礎(chǔ)上terminated()方法也業(yè)已執(zhí)行完畢，才會從上個狀態(tài)進(jìn)入這個狀態(tài)，代表線程池已經(jīng)完全停止。

由于線程池的狀態(tài)都是通過AtomicInteger來保存的，可以通過比較的方式簡單的得到當(dāng)前線程狀態(tài)。

private final BlockingQueue<Runnable> workQueue;?
private final ReentrantLock mainLock = new ReentrantLock();?
private final HashSet<Worker> workers = new HashSet<Worker>();?
private final Condition termination = mainLock.newCondition();?
private int largestPoolSize;?
private long completedTaskCount;?
private volatile ThreadFactory threadFactory;?
private volatile RejectedExecutionHandler handler;?
private volatile long keepAliveTime;?
private volatile boolean allowCoreThreadTimeOut;?
private volatile int corePoolSize;?
private volatile int maximumPoolSize;

接下來是線程池的幾個有關(guān)工作線程的變量

corePoolSize表示線程池中允許存活最少的工作線程數(shù)量，但值得注意的是如果allowCoreThreadTimeOut一旦設(shè)置true(默認(rèn)false),每個線程的存活時間只有keepAliveTime也就是說在allowCoreThreadTimeOut為true的時候，該線程池最小的工作線程數(shù)量為0；maximumPoolSize代表線程池中最大的工作線程數(shù)量。
keepAliveTime為線程池中工作線程數(shù)量大于corePoolSize時，每個工作線程的在等待工作時最長的等待時間。
workQueue作為線程池的任務(wù)等待隊列，這個將在接下來的execute()里詳細(xì)解釋。
Workers作為存放線程池中存放工作線程的容器。
largestPoolSize用來記錄線程池中存在過的最大的工作線程數(shù)量。
completedTaskCount用來記錄線程池完成的任務(wù)的總數(shù)。
Handler作為線程池中在不能接受任務(wù)的時候的拒絕策略，我們可以實現(xiàn)自己的拒絕策略，在實現(xiàn)了RejectedExecutionHandler接口的前提下。下面是線程池的默認(rèn)拒絕策略，

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
? ? throw new RejectedExecutionException("Task " + r.toString() +
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?" rejected from " +
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?e.toString());
}

threadFactory作為線程池生產(chǎn)線程的工廠類

下面是線程池默認(rèn)的線程工廠的生產(chǎn)線程方法

public Thread newThread(Runnable r) {
? ? Thread t = new Thread(group, r,
? ? ? ? ? ? ? ? ? ? ? ? ? namePrefix + threadNumber.getAndIncrement(),
? ? ? ? ? ? ? ? ? ? ? ? ? 0);
? ? if (t.isDaemon())
? ? ? ? t.setDaemon(false);
? ? if (t.getPriority() != Thread.NORM_PRIORITY)
? ? ? ? t.setPriority(Thread.NORM_PRIORITY);
? ? return t;
}

我們可以先看我們最常調(diào)用的execute()方法

public void execute(Runnable command) {
? ? if (command == null)
? ? ? ? throw new NullPointerException();
?
? ? int c = ctl.get();
? ? if (workerCountOf(c) < corePoolSize) {
? ? ? ? if (addWorker(command, true))
? ? ? ? ? ? return;
? ? ? ? c = ctl.get();
? ? }
? ? if (isRunning(c) && workQueue.offer(command)) {
? ? ? ? int recheck = ctl.get();
? ? ? ? if (! isRunning(recheck) && remove(command))
? ? ? ? ? ? reject(command);
? ? ? ? else if (workerCountOf(recheck) == 0)
? ? ? ? ? ? addWorker(null, false);
? ? }
? ? else if (!addWorker(command, false))
? ? ? ? reject(command);
}

execute()內(nèi)部的調(diào)用邏輯非常清晰。

如果當(dāng)前線程池的工作線程數(shù)量小于corePoolSize，那么直接調(diào)用addWoker()，來添加工作線程。

下面是addWorker()的具體方法

private boolean addWorker(Runnable firstTask, boolean core) {
? ? retry:
? ? for (;;) {
? ? ? ? int c = ctl.get();
? ? ? ? int rs = runStateOf(c);
? ? ? ? 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;
? ? ? ? }
? ? }
?
? ? boolean workerStarted = false;
? ? boolean workerAdded = false;
? ? Worker w = null;
? ? try {
? ? ? ? final ReentrantLock mainLock = this.mainLock;
? ? ? ? w = new Worker(firstTask);
? ? ? ? final Thread t = w.thread;
? ? ? ? if (t != null) {
? ? ? ? ? ? mainLock.lock();
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? int c = ctl.get();
? ? ? ? ? ? ? ? int rs = runStateOf(c);
?
? ? ? ? ? ? ? ? if (rs < SHUTDOWN ||
? ? ? ? ? ? ? ? ? ? (rs == SHUTDOWN && firstTask == null)) {
? ? ? ? ? ? ? ? ? ? if (t.isAlive()) ? ? ? ? ? ? ? ? ? ? ? ??
?? ??? ? ? 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;
}

這段方法比較長，但整體的邏輯還是清晰的。

首先判斷當(dāng)前線程池的狀態(tài)，如果已經(jīng)狀態(tài)不是shutdown或者running，或者已經(jīng)為shutdown但是工作隊列已經(jīng)為空，那么這個時候直接返回添加工作失敗。接下來是對線程池線程數(shù)量的判斷，根據(jù)調(diào)用時的core的值來判斷是跟corePoolSize還是 maximumPoolSize判斷。

在確認(rèn)了線程池狀態(tài)以及線程池中工作線程數(shù)量之后，才真正開始添加工作線程。

新建立一個worker類（線程池的內(nèi)部類，具體的工作線程），將要執(zhí)行的具體線程做為構(gòu)造方法中的參數(shù)傳遞進(jìn)去，接下來將其加入線程池的工作線程容器workers，并且更新工作線程最大量，最后調(diào)用worker工作線程的start()方法，就完成了工作線程的建立與啟動。

讓我們回到execute()方法，如果我們在一開始的線程數(shù)量就大于corePoolSize，或者我們在調(diào)用addworker()方法的過程中出現(xiàn)了問題導(dǎo)致添加工作線程數(shù)量失敗，那么我們會繼續(xù)執(zhí)行接下來的邏輯。

在判斷完畢線程池的狀態(tài)后，則會將任務(wù)通過workQueue.offer())方法試圖加進(jìn)任務(wù)隊列。Offer()方法的具體實現(xiàn)會根據(jù)在線程池構(gòu)造方法中選取的任務(wù)隊列種類而產(chǎn)生變化。

但是如果成功加入了任務(wù)隊列，仍舊需要注意判斷如果線程池的狀態(tài)如果已經(jīng)不是running那么會拒絕執(zhí)行這一任務(wù)并執(zhí)行相應(yīng)的拒絕策略。在最后需要記得成功加入隊列成功后如果線程池中如果已經(jīng)沒有了工作線程，需要重新建立一個工作線程去執(zhí)行仍舊在任務(wù)隊列中等待執(zhí)行的任務(wù)。

如果在之前的前提下加入任務(wù)隊列也失敗了（比如任務(wù)隊列已滿），則會在不超過線程池最大線程數(shù)量的前提下建立一個工作線程來處理。

如果在最后的建立工作線程也失敗了，那么我們只有很遺憾的執(zhí)行任務(wù)的拒絕策略了。

在之前的過程中我們建立了工作線程Worker()類，那么我們現(xiàn)在看看worker類的內(nèi)部實現(xiàn)，也可以說是線程池的核心部分。

Worker類作為線程池的內(nèi)部類

接下來是Worker()類的成員

final Thread thread;
?
Runnable firstTask;
?
volatile long completedTasks;

thread作為worker的工作線程空間，由線程池中所設(shè)置的線程工廠生成。

firstTask則是worker在構(gòu)造方法中所接受到的所要執(zhí)行的任務(wù)。
completedTasks作為該worker類所執(zhí)行完畢的任務(wù)總數(shù)。

接下來我們可以看最重要的，也就是我們之前建立完Worker類之后立馬調(diào)用的run()方法了

public void run() {
? ? runWorker(this);
}

run()方法實現(xiàn)的很簡單

我們可以繼續(xù)追蹤下去

final void runWorker(Worker w) {
? ? Thread wt = Thread.currentThread();
? ? Runnable task = w.firstTask;
? ? w.firstTask = null;
? ? w.unlock();?
? ? boolean completedAbruptly = true;
? ? try {
? ? ? ? while (task != null || (task = getTask()) != null) {
? ? ? ? ? ? w.lock();
? ? ? ? ? ? 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);
? ? }
}

如果這個worker還沒有執(zhí)行過在構(gòu)造方法就傳入的任務(wù)，那么在這個方法中，會直接執(zhí)行這一任務(wù)，如果沒有，則會嘗試去從任務(wù)隊列當(dāng)中去取的新的任務(wù)。

但是在真正調(diào)用任務(wù)之前，仍舊會判斷線程池的狀態(tài)，如果已經(jīng)不是running亦或是shutdwon，則會直接確保線程被中斷。如果沒有，將會繼續(xù)執(zhí)行并確保不被中斷。

接下來可見，我們所需要的任務(wù)，直接在工作線程中直接以run()方式以非線程的方式所調(diào)用，這里也就是我們所需要的任務(wù)真正執(zhí)行的地方。

在執(zhí)行完畢后，工作線程的使命并沒有真正宣告段落。在while部分worker仍舊會通過getTask()方法試圖取得新的任務(wù)。

下面是getTask()的實現(xiàn)

private Runnable getTask() {
? ? boolean timedOut = false;?
? ? retry:
? ? for (;;) {
? ? ? ? int c = ctl.get();
? ? ? ? int rs = runStateOf(c);
?
? ? ? ? ? ? ? ?if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
? ? ? ? ? ? decrementWorkerCount();
? ? ? ? ? ? return null;
? ? ? ? }
?
? ? ? ? boolean timed; ? ? ? ? ? ?
? ? ? ? for (;;) {
? ? ? ? ? ? int wc = workerCountOf(c);
? ? ? ? ? ? timed = allowCoreThreadTimeOut || wc > corePoolSize;
?
? ? ? ? ? ? if (wc <= maximumPoolSize && ! (timedOut && timed))
? ? ? ? ? ? ? ? break;
? ? ? ? ? ? if (compareAndDecrementWorkerCount(c))
? ? ? ? ? ? ? ? return null;
? ? ? ? ? ? c = ctl.get(); ?
? ? ? ? ? ? if (runStateOf(c) != rs)
? ? ? ? ? ? ? ? continue retry;
? ? ? ? }
?
? ? ? ? try {
? ? ? ? ? ? Runnable r = timed ?
? ? ? ? ? ? ? ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
? ? ? ? ? ? ? ? workQueue.take();
? ? ? ? ? ? if (r != null)
? ? ? ? ? ? ? ? return r;
? ? ? ? ? ? timedOut = true;
? ? ? ? } catch (InterruptedException retry) {
? ? ? ? ? ? timedOut = false;
? ? ? ? }
? ? }
}

首先仍舊會判斷線程池的狀態(tài)是否是running還是shutdown以及stop狀態(tài)下隊列是否仍舊有需要等待執(zhí)行的任務(wù)。如果狀態(tài)沒有問題，則會跟據(jù)allowCoreThreadTimeOut和corePoolSize的值通過對前面這兩個屬性解釋的方式來選擇從任務(wù)隊列中獲得任務(wù)的方式（是否設(shè)置timeout）。其中的timedOut保證了確認(rèn)前一次試圖取任務(wù)時超時發(fā)生的記錄，以確保工作線程的回收。

在runWorker()方法的最后

調(diào)用了processWorkerExist()方法來執(zhí)行工作線程的回收。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
? ? if (completedAbruptly)?
? ? ? ? decrementWorkerCount();
?
? ? final ReentrantLock mainLock = this.mainLock;
? ? mainLock.lock();
? ? try {
? ? ? ? completedTaskCount += w.completedTasks;
? ? ? ? workers.remove(w);
? ? } finally {
? ? ? ? mainLock.unlock();
? ? }
?
? ? tryTerminate();?
? ? int c = ctl.get();
? ? if (runStateLessThan(c, STOP)) {
? ? ? ? if (!completedAbruptly) {
? ? ? ? ? ? int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
? ? ? ? ? ? if (min == 0 && ! workQueue.isEmpty())
? ? ? ? ? ? ? ? min = 1;
? ? ? ? ? ? if (workerCountOf(c) >= min)
? ? ? ? ? ? ? ? return;?
? ? ? ? }
? ? ? ? addWorker(null, false);
? ? }
}

在這一方法中，首先確保已經(jīng)重新更新了線程池中工作線程的數(shù)量，之后從線程池中的工作線程容器移去當(dāng)前工作線程，并且將完成的任務(wù)總數(shù)加到線程池的任務(wù)總數(shù)當(dāng)中。

在最后仍舊要確保線程池中依舊存在大于等于最小線程數(shù)量的工作線程數(shù)量存在，如果沒有，則重新建立工作線程去等待處理任務(wù)隊列中任務(wù)。

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: