java高并發(fā)ThreadPoolExecutor類(lèi)解析線(xiàn)程池執(zhí)行流程

更新時(shí)間：2022年09月27日 16:19:52 作者：冰河

這篇文章主要為大家介紹了java高并發(fā)ThreadPoolExecutor類(lèi)解析線(xiàn)程池執(zhí)行流程，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

摘要

ThreadPoolExecutor是Java線(xiàn)程池中最核心的類(lèi)之一，它能夠保證線(xiàn)程池按照正常的業(yè)務(wù)邏輯執(zhí)行任務(wù)，并通過(guò)原子方式更新線(xiàn)程池每個(gè)階段的狀態(tài)。

今天，我們通過(guò)ThreadPoolExecutor類(lèi)的源碼深度解析線(xiàn)程池執(zhí)行任務(wù)的核心流程，小伙伴們最好是打開(kāi)IDEA，按照冰河說(shuō)的步驟，調(diào)試下ThreadPoolExecutor類(lèi)的源碼，這樣會(huì)理解的更加深刻，好了，開(kāi)始今天的主題。

核心邏輯概述

ThreadPoolExecutor類(lèi)中存在一個(gè)workers工作線(xiàn)程集合，用戶(hù)可以向線(xiàn)程池中添加需要執(zhí)行的任務(wù)，workers集合中的工作線(xiàn)程可以直接執(zhí)行任務(wù)，或者從任務(wù)隊(duì)列中獲取任務(wù)后執(zhí)行。ThreadPoolExecutor類(lèi)中提供了整個(gè)線(xiàn)程池從創(chuàng)建到執(zhí)行任務(wù)，再到消亡的整個(gè)流程方法。本文，就結(jié)合ThreadPoolExecutor類(lèi)的源碼深度分析線(xiàn)程池執(zhí)行任務(wù)的整體流程。

在ThreadPoolExecutor類(lèi)中，線(xiàn)程池的邏輯主要體現(xiàn)在execute(Runnable)方法，addWorker(Runnable, boolean)方法，addWorkerFailed(Worker)方法和拒絕策略上，接下來(lái)，我們就深入分析這幾個(gè)核心方法。

execute(Runnable)方法

execute(Runnable)方法的作用是提交Runnable類(lèi)型的任務(wù)到線(xiàn)程池中。我們先看下execute(Runnable)方法的源碼，如下所示。

public void execute(Runnable command) {
	//如果提交的任務(wù)為空，則拋出空指針異常
	if (command == null)
		throw new NullPointerException();
	//獲取線(xiàn)程池的狀態(tài)和線(xiàn)程池中線(xiàn)程的數(shù)量
	int c = ctl.get();
	//線(xiàn)程池中的線(xiàn)程數(shù)量小于corePoolSize的值
	if (workerCountOf(c) < corePoolSize) {
		//重新開(kāi)啟線(xiàn)程執(zhí)行任務(wù)
		if (addWorker(command, true))
			return;
		c = ctl.get();
	}
	//如果線(xiàn)程池處于RUNNING狀態(tài)，則將任務(wù)添加到阻塞隊(duì)列中
	if (isRunning(c) && workQueue.offer(command)) {
		//再次獲取線(xiàn)程池的狀態(tài)和線(xiàn)程池中線(xiàn)程的數(shù)量，用于二次檢查
		int recheck = ctl.get();
		//如果線(xiàn)程池沒(méi)有未處于RUNNING狀態(tài)，從隊(duì)列中刪除任務(wù)
		if (! isRunning(recheck) && remove(command))
			//執(zhí)行拒絕策略
			reject(command);
		//如果線(xiàn)程池為空，則向線(xiàn)程池中添加一個(gè)線(xiàn)程
		else if (workerCountOf(recheck) == 0)
			addWorker(null, false);
	}
	//任務(wù)隊(duì)列已滿(mǎn)，則新增worker線(xiàn)程，如果新增線(xiàn)程失敗，則執(zhí)行拒絕策略
	else if (!addWorker(command, false))
		reject(command);
}

整個(gè)任務(wù)的執(zhí)行流程，我們可以簡(jiǎn)化成下圖所示。

接下來(lái)，我們拆解execute(Runnable)方法，具體分析execute(Runnable)方法的執(zhí)行邏輯。

（1）線(xiàn)程池中的線(xiàn)程數(shù)是否小于corePoolSize核心線(xiàn)程數(shù)，如果小于corePoolSize核心線(xiàn)程數(shù)，則向workers工作線(xiàn)程集合中添加一個(gè)核心線(xiàn)程執(zhí)行任務(wù)。代碼如下所示。

//線(xiàn)程池中的線(xiàn)程數(shù)量小于corePoolSize的值
if (workerCountOf(c) < corePoolSize) {
	//重新開(kāi)啟線(xiàn)程執(zhí)行任務(wù)
	if (addWorker(command, true))
		return;
	c = ctl.get();
}

（2）如果線(xiàn)程池中的線(xiàn)程數(shù)量大于corePoolSize核心線(xiàn)程數(shù)，則判斷當(dāng)前線(xiàn)程池是否處于RUNNING狀態(tài)，如果處于RUNNING狀態(tài)，則添加任務(wù)到待執(zhí)行的任務(wù)隊(duì)列中。注意：這里向任務(wù)隊(duì)列添加任務(wù)時(shí)，需要判斷線(xiàn)程池是否處于RUNNING狀態(tài)，只有線(xiàn)程池處于RUNNING狀態(tài)時(shí)，才能向任務(wù)隊(duì)列添加新任務(wù)。否則，會(huì)執(zhí)行拒絕策略。代碼如下所示。

if (isRunning(c) && workQueue.offer(command))

（3）向任務(wù)隊(duì)列中添加任務(wù)成功，由于其他線(xiàn)程可能會(huì)修改線(xiàn)程池的狀態(tài)，所以這里需要對(duì)線(xiàn)程池進(jìn)行二次檢查，如果當(dāng)前線(xiàn)程池的狀態(tài)不再是RUNNING狀態(tài)，則需要將添加的任務(wù)從任務(wù)隊(duì)列中移除，執(zhí)行后續(xù)的拒絕策略。如果當(dāng)前線(xiàn)程池仍然處于RUNNING狀態(tài)，則判斷線(xiàn)程池是否為空，如果線(xiàn)程池中不存在任何線(xiàn)程，則新建一個(gè)線(xiàn)程添加到線(xiàn)程池中，如下所示。

//再次獲取線(xiàn)程池的狀態(tài)和線(xiàn)程池中線(xiàn)程的數(shù)量，用于二次檢查
int recheck = ctl.get();
//如果線(xiàn)程池沒(méi)有未處于RUNNING狀態(tài)，從隊(duì)列中刪除任務(wù)
if (! isRunning(recheck) && remove(command))
	//執(zhí)行拒絕策略
	reject(command);
//如果線(xiàn)程池為空，則向線(xiàn)程池中添加一個(gè)線(xiàn)程
else if (workerCountOf(recheck) == 0)
	addWorker(null, false);

（4）如果在步驟（3）中向任務(wù)隊(duì)列中添加任務(wù)失敗，則嘗試開(kāi)啟新的線(xiàn)程執(zhí)行任務(wù)。此時(shí)，如果線(xiàn)程池中的線(xiàn)程數(shù)量已經(jīng)大于線(xiàn)程池中的最大線(xiàn)程數(shù)maximumPoolSize，則不能再啟動(dòng)新線(xiàn)程。此時(shí)，表示線(xiàn)程池中的任務(wù)隊(duì)列已滿(mǎn)，并且線(xiàn)程池中的線(xiàn)程已滿(mǎn)，需要執(zhí)行拒絕策略，代碼如下所示。

//任務(wù)隊(duì)列已滿(mǎn)，則新增worker線(xiàn)程，如果新增線(xiàn)程失敗，則執(zhí)行拒絕策略
else if (!addWorker(command, false))
	reject(command);

這里，我們將execute(Runnable)方法拆解，結(jié)合流程圖來(lái)理解線(xiàn)程池中任務(wù)的執(zhí)行流程就比較簡(jiǎn)單了。可以這么說(shuō)，execute(Runnable)方法的邏輯基本上就是一般線(xiàn)程池的執(zhí)行邏輯，理解了execute(Runnable)方法，就基本理解了線(xiàn)程池的執(zhí)行邏輯。

注意：有關(guān)ScheduledThreadPoolExecutor類(lèi)和ForkJoinPool類(lèi)執(zhí)行線(xiàn)程池的邏輯，在【高并發(fā)專(zhuān)題】系列文章中的后文中會(huì)詳細(xì)說(shuō)明，理解了這些類(lèi)的執(zhí)行邏輯，就基本全面掌握了線(xiàn)程池的執(zhí)行流程。

在分析execute(Runnable)方法的源碼時(shí)，我們發(fā)現(xiàn)execute(Runnable)方法中多處調(diào)用了addWorker(Runnable, boolean)方法，接下來(lái)，我們就一起分析下addWorker(Runnable, boolean)方法的邏輯。

addWorker(Runnable, boolean)方法

總體上，addWorker(Runnable, boolean)方法可以分為三部分，第一部分是使用CAS安全的向線(xiàn)程池中添加工作線(xiàn)程；第二部分是創(chuàng)建新的工作線(xiàn)程；第三部分則是將任務(wù)通過(guò)安全的并發(fā)方式添加到workers中，并啟動(dòng)工作線(xiàn)程執(zhí)行任務(wù)。

接下來(lái)，我們看下addWorker(Runnable, boolean)方法的源碼，如下所示。

private boolean addWorker(Runnable firstTask, boolean core) {
	//標(biāo)記重試的標(biāo)識(shí)
	retry:
	for (;;) {
		int c = ctl.get();
		int rs = runStateOf(c);
		// 檢查隊(duì)列是否在某些特定的條件下為空
		if (rs >= SHUTDOWN &&
			! (rs == SHUTDOWN &&
			   firstTask == null &&
			   ! workQueue.isEmpty()))
			return false;
		//下面循環(huán)的主要作用為通過(guò)CAS方式增加線(xiàn)程的個(gè)數(shù)
		for (;;) {
			//獲取線(xiàn)程池中的線(xiàn)程數(shù)量
			int wc = workerCountOf(c);
			//如果線(xiàn)程池中的線(xiàn)程數(shù)量超出限制，直接返回false
			if (wc >= CAPACITY ||
				wc >= (core ? corePoolSize : maximumPoolSize))
				return false;
			//通過(guò)CAS方式向線(xiàn)程池新增線(xiàn)程數(shù)量
			if (compareAndIncrementWorkerCount(c))
				//通過(guò)CAS方式保證只有一個(gè)線(xiàn)程執(zhí)行成功，跳出最外層循環(huán)
				break retry;
			//重新獲取ctl的值
			c = ctl.get();  
			//如果CAS操作失敗了，則需要在內(nèi)循環(huán)中重新嘗試通過(guò)CAS新增線(xiàn)程數(shù)量
			if (runStateOf(c) != rs)
				continue retry;
		}
	}
	//跳出最外層for循環(huán)，說(shuō)明通過(guò)CAS新增線(xiàn)程數(shù)量成功
	//此時(shí)創(chuàng)建新的工作線(xiàn)程
	boolean workerStarted = false;
	boolean workerAdded = false;
	Worker w = null;
	try {
		//將執(zhí)行的任務(wù)封裝成worker
		w = new Worker(firstTask);
		final Thread t = w.thread;
		if (t != null) {
			//獨(dú)占鎖，保證操作workers時(shí)的同步
			final ReentrantLock mainLock = this.mainLock;
			mainLock.lock();
			try {
				//此處需要重新檢查線(xiàn)程池狀態(tài)
				//原因是在獲得鎖之前可能其他的線(xiàn)程改變了線(xiàn)程池的狀態(tài)
				int rs = runStateOf(ctl.get());
				if (rs < SHUTDOWN ||
					(rs == SHUTDOWN && firstTask == null)) {
					if (t.isAlive())
						throw new IllegalThreadStateException();
					//向worker中添加新任務(wù)
					workers.add(w);
					int s = workers.size();
					if (s > largestPoolSize)
						largestPoolSize = s;
					//將是否添加了新任務(wù)的標(biāo)識(shí)設(shè)置為true
					workerAdded = true;
				}
			} finally {
				//釋放獨(dú)占鎖
				mainLock.unlock();
			}
			//添加新任成功，則啟動(dòng)線(xiàn)程執(zhí)行任務(wù)
			if (workerAdded) {
				t.start();
				//將任務(wù)是否已經(jīng)啟動(dòng)的標(biāo)識(shí)設(shè)置為true
				workerStarted = true;
			}
		}
	} finally {
		//如果任務(wù)未啟動(dòng)或啟動(dòng)失敗，則調(diào)用addWorkerFailed(Worker)方法
		if (! workerStarted)
			addWorkerFailed(w);
	}
	//返回是否啟動(dòng)任務(wù)的標(biāo)識(shí)
	return workerStarted;
}

乍一看，addWorker(Runnable, boolean)方法還蠻長(zhǎng)的，這里，我們還是將addWorker(Runnable, boolean)方法進(jìn)行拆解。

（1）檢查任務(wù)隊(duì)列是否在某些特定的條件下為空，代碼如下所示。

// 檢查隊(duì)列是否在某些特定的條件下為空
if (rs >= SHUTDOWN &&
	! (rs == SHUTDOWN &&
	   firstTask == null &&
	   ! workQueue.isEmpty()))
	return false;

（2）在通過(guò)步驟（1）的校驗(yàn)后，則進(jìn)入內(nèi)層for循環(huán)，在內(nèi)層for循環(huán)中通過(guò)CAS來(lái)增加線(xiàn)程池中的線(xiàn)程數(shù)量，如果CAS操作成功，則直接退出雙重for循環(huán)。如果CAS操作失敗，則查看當(dāng)前線(xiàn)程池的狀態(tài)是否發(fā)生了變化，如果線(xiàn)程池的狀態(tài)發(fā)生了變化，則通過(guò)continue關(guān)鍵字重新通過(guò)外層for循環(huán)校驗(yàn)任務(wù)隊(duì)列，檢驗(yàn)通過(guò)再次執(zhí)行內(nèi)層for循環(huán)的CAS操作。如果線(xiàn)程池的狀態(tài)沒(méi)有發(fā)生變化，此時(shí)上一次CAS操作失敗了，則繼續(xù)嘗試CAS操作。代碼如下所示。

for (;;) {
	//獲取線(xiàn)程池中的線(xiàn)程數(shù)量
	int wc = workerCountOf(c);
	//如果線(xiàn)程池中的線(xiàn)程數(shù)量超出限制，直接返回false
	if (wc >= CAPACITY ||
		wc >= (core ? corePoolSize : maximumPoolSize))
		return false;
	//通過(guò)CAS方式向線(xiàn)程池新增線(xiàn)程數(shù)量
	if (compareAndIncrementWorkerCount(c))
		//通過(guò)CAS方式保證只有一個(gè)線(xiàn)程執(zhí)行成功，跳出最外層循環(huán)
		break retry;
	//重新獲取ctl的值
	c = ctl.get();  
	//如果CAS操作失敗了，則需要在內(nèi)循環(huán)中重新嘗試通過(guò)CAS新增線(xiàn)程數(shù)量
	if (runStateOf(c) != rs)
		continue retry;
}

（3）CAS操作成功后，表示向線(xiàn)程池中成功添加了工作線(xiàn)程，此時(shí)，還沒(méi)有線(xiàn)程去執(zhí)行任務(wù)。使用全局的獨(dú)占鎖mainLock來(lái)將新增的工作線(xiàn)程Worker對(duì)象安全的添加到workers中。

總體邏輯就是：創(chuàng)建新的Worker對(duì)象，并獲取Worker對(duì)象中的執(zhí)行線(xiàn)程，如果線(xiàn)程不為空，則獲取獨(dú)占鎖，獲取鎖成功后，再次檢查線(xiàn)線(xiàn)程的狀態(tài)，這是避免在獲取獨(dú)占鎖之前其他線(xiàn)程修改了線(xiàn)程池的狀態(tài)，或者關(guān)閉了線(xiàn)程池。如果線(xiàn)程池關(guān)閉，則需要釋放鎖。否則將新增加的線(xiàn)程添加到工作集合中，釋放鎖并啟動(dòng)線(xiàn)程執(zhí)行任務(wù)。將是否啟動(dòng)線(xiàn)程的標(biāo)識(shí)設(shè)置為true。最后，判斷線(xiàn)程是否啟動(dòng)，如果沒(méi)有啟動(dòng)，則調(diào)用addWorkerFailed(Worker)方法。最終返回線(xiàn)程是否起送的標(biāo)識(shí)。

//跳出最外層for循環(huán)，說(shuō)明通過(guò)CAS新增線(xiàn)程數(shù)量成功
//此時(shí)創(chuàng)建新的工作線(xiàn)程
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
	//將執(zhí)行的任務(wù)封裝成worker
	w = new Worker(firstTask);
	final Thread t = w.thread;
	if (t != null) {
		//獨(dú)占鎖，保證操作workers時(shí)的同步
		final ReentrantLock mainLock = this.mainLock;
		mainLock.lock();
		try {
			//此處需要重新檢查線(xiàn)程池狀態(tài)
			//原因是在獲得鎖之前可能其他的線(xiàn)程改變了線(xiàn)程池的狀態(tài)
			int rs = runStateOf(ctl.get());
			if (rs < SHUTDOWN ||
				(rs == SHUTDOWN && firstTask == null)) {
				if (t.isAlive())
					throw new IllegalThreadStateException();
				//向worker中添加新任務(wù)
				workers.add(w);
				int s = workers.size();
				if (s > largestPoolSize)
					largestPoolSize = s;
				//將是否添加了新任務(wù)的標(biāo)識(shí)設(shè)置為true
				workerAdded = true;
			}
		} finally {
			//釋放獨(dú)占鎖
			mainLock.unlock();
		}
		//添加新任成功，則啟動(dòng)線(xiàn)程執(zhí)行任務(wù)
		if (workerAdded) {
			t.start();
			//將任務(wù)是否已經(jīng)啟動(dòng)的標(biāo)識(shí)設(shè)置為true
			workerStarted = true;
		}
	}
} finally {
	//如果任務(wù)未啟動(dòng)或啟動(dòng)失敗，則調(diào)用addWorkerFailed(Worker)方法
	if (! workerStarted)
		addWorkerFailed(w);
}
//返回是否啟動(dòng)任務(wù)的標(biāo)識(shí)
return workerStarted;

addWorkerFailed(Worker)方法

在addWorker(Runnable, boolean)方法中，如果添加工作線(xiàn)程失敗或者工作線(xiàn)程啟動(dòng)失敗時(shí)，則會(huì)調(diào)用addWorkerFailed(Worker)方法，下面我們就來(lái)看看addWorkerFailed(Worker)方法的實(shí)現(xiàn)，如下所示。

private void addWorkerFailed(Worker w) {
	//獲取獨(dú)占鎖
	final ReentrantLock mainLock = this.mainLock;
	mainLock.lock();
	try {
		//如果Worker任務(wù)不為空
		if (w != null)
			//將任務(wù)從workers集合中移除
			workers.remove(w);
		//通過(guò)CAS將任務(wù)數(shù)量減1
		decrementWorkerCount();
		tryTerminate();
	} finally {
		//釋放鎖
		mainLock.unlock();
	}
}

addWorkerFailed(Worker)方法的邏輯就比較簡(jiǎn)單了，獲取獨(dú)占鎖，將任務(wù)從workers中移除，并且通過(guò)CAS將任務(wù)的數(shù)量減1，最后釋放鎖。

拒絕策略

我們?cè)诜治鰁xecute(Runnable)方法時(shí)，線(xiàn)程池會(huì)在適當(dāng)?shù)臅r(shí)候調(diào)用reject(Runnable)方法來(lái)執(zhí)行相應(yīng)的拒絕策略，我們看下reject(Runnable)方法的實(shí)現(xiàn)，如下所示。

final void reject(Runnable command) {
	handler.rejectedExecution(command, this);
}

通過(guò)代碼，我們發(fā)現(xiàn)調(diào)用的是handler的rejectedExecution方法，handler又是個(gè)什么鬼，我們繼續(xù)跟進(jìn)代碼，如下所示。

private volatile RejectedExecutionHandler handler;

再看看RejectedExecutionHandler是個(gè)啥類(lèi)型，如下所示。

package java.util.concurrent;
public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

可以發(fā)現(xiàn)RejectedExecutionHandler是個(gè)接口，定義了一個(gè)rejectedExecution(Runnable, ThreadPoolExecutor)方法。既然RejectedExecutionHandler是個(gè)接口，那我們就看看有哪些類(lèi)實(shí)現(xiàn)了RejectedExecutionHandler接口。

看到這里，我們發(fā)現(xiàn)RejectedExecutionHandler接口的實(shí)現(xiàn)類(lèi)正是線(xiàn)程池默認(rèn)提供的四種拒絕策略的實(shí)現(xiàn)類(lèi)。

至于reject(Runnable)方法中具體會(huì)執(zhí)行哪個(gè)類(lèi)的拒絕策略，是根據(jù)創(chuàng)建線(xiàn)程池時(shí)傳遞的參數(shù)決定的。如果沒(méi)有傳遞拒絕策略，則默認(rèn)會(huì)執(zhí)行AbortPolicy類(lèi)的拒絕策略。否則會(huì)執(zhí)行傳遞的類(lèi)的拒絕策略。

在創(chuàng)建線(xiàn)程池時(shí)，除了能夠傳遞JDK默認(rèn)提供的拒絕策略外，還可以傳遞自定義的拒絕策略。如果想使用自定義的拒絕策略，則只需要實(shí)現(xiàn)RejectedExecutionHandler接口，并重寫(xiě)rejectedExecution(Runnable, ThreadPoolExecutor)方法即可。例如，下面的代碼。

public class CustomPolicy implements RejectedExecutionHandler {
	public CustomPolicy() { }
	public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
		if (!e.isShutdown()) {
			System.out.println("使用調(diào)用者所在的線(xiàn)程來(lái)執(zhí)行任務(wù)")
			r.run();
		}
	}
}

使用如下方式創(chuàng)建線(xiàn)程池。

new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                       60L, TimeUnit.SECONDS,
                       new SynchronousQueue<Runnable>(),
                       Executors.defaultThreadFactory(),
		       new CustomPolicy());

至此，線(xiàn)程池執(zhí)行任務(wù)的整體核心邏輯分析結(jié)束，更多關(guān)于java ThreadPoolExecutor類(lèi)解析線(xiàn)程池的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: