先說說使用線程池的好處，比如可以控制線程的數(shù)量，節(jié)省反復(fù)創(chuàng)建線程和銷毀線程的開銷等，在開發(fā)中的使用，一般來說任務(wù)量肯定是大于線程數(shù)量的，而為了防止出現(xiàn)OOM，都是建議設(shè)置相對應(yīng)業(yè)務(wù)的合適線程數(shù)量。那是在線程池中線程就只有那些，肯定是要做到線程的重復(fù)利用，才能執(zhí)行超過線程的任務(wù)量的，那么線程池是怎么做到"線程復(fù)用"的呢？

從源碼開始逐步分析，從execute()方法開始

//原子變量 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));public void execute(Runnable command) {//判斷提交的Runnable 任務(wù)，如果為null，則報(bào)NullPointerExceptionif (command == null)throw new NullPointerException(); int c = ctl.get();//判斷當(dāng)前線程數(shù)是否小于核心線程數(shù)，如果小于，那就調(diào)用addWorker方法新增一個(gè)Worker，也可以理解成一個(gè)線程if (workerCountOf(c) < corePoolSize) {//addWorker這個(gè)方法主要要做的事就是執(zhí)行command，同時(shí)第二個(gè)參數(shù)決定以哪個(gè)界限來進(jìn)行是否新增線程的判斷，傳入true則代表以核心線程數(shù)為判斷條件if (addWorker(command, true))return;c = ctl.get();}//走到這步邏輯，則說明線程數(shù)大于等于核心線程數(shù)，或者addWorker方法調(diào)用失敗了，這時(shí)就判斷線程池是否是Running狀態(tài)，如果是就調(diào)用offer方法提交線程任務(wù)到任務(wù)隊(duì)列中if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();//如果線程池狀態(tài)不是Running，說明線程池已經(jīng)被關(guān)閉，這時(shí)就移除新提交到隊(duì)列中的任務(wù)if (! isRunning(recheck) && remove(command))//執(zhí)行拒絕策略reject(command);//檢查下當(dāng)前線程數(shù)是不是為0，為0的話就沒有線程執(zhí)行任務(wù)了 else if (workerCountOf(recheck) == 0)//所以就通過addWorker新建一個(gè)線程addWorker(null, false);}//走到這步邏輯，要么是線程池狀態(tài)不是Running，說明已經(jīng)關(guān)閉了，要么就是添加任務(wù)進(jìn)任務(wù)隊(duì)列時(shí)失敗了，說明任務(wù)隊(duì)列滿了，這時(shí)候就該添加最大線程數(shù)了，傳入false則代表以最大線程數(shù)為判斷條件else if (!addWorker(command, false))//以上addWorker方法如果返回結(jié)果是false，就會執(zhí)行拒絕策略了reject(command);}

以上粗略的過了下execute()方法的源碼，可以發(fā)現(xiàn)有個(gè)addWorker()方法無處不在，同時(shí)它也有新建線程和運(yùn)行任務(wù)的作用，所以重點(diǎn)看下

addWorker()方法源碼?

private boolean addWorker(Runnable firstTask, boolean core) {//可以當(dāng)做一個(gè)標(biāo)記位retry://沒有主動跳出就會無限的循環(huán)for (;;) {int c = ctl.get();int rs = runStateOf(c);//做校驗(yàn)，比如rs >= SHUTDOWN說明線程池狀態(tài)不正常，比如rs == SHUTDOWN說明線程池關(guān)閉了， firstTask == null說明無任務(wù)執(zhí)行等，就直接返回falseif (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;//沒有主動跳出就會無限的循環(huán)for (;;) {//獲取當(dāng)前線程數(shù)int wc = workerCountOf(c);//判斷下當(dāng)前線程數(shù)是不是超過規(guī)定了，是就直接返回falseif (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;//CAS操作，比較并做+1操作if (compareAndIncrementWorkerCount(c))//失敗了的話就直接跳出循環(huán)到retry那里，且不再進(jìn)循環(huán)了break retry; c = ctl.get(); // Re-read ctl// 如果線程池狀態(tài)被更改的話if (runStateOf(c) != rs)//直接跳出循環(huán)到retry那里，會再次進(jìn)循環(huán)continue retry;// else CAS failed due to workerCount change; retry inner loop}}boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {//創(chuàng)建一個(gè)Worker對象，以Runnable實(shí)例對象firstTask為參數(shù)w = new Worker(firstTask);//拿到線程對象final Thread t = w.thread;//線程對象不為空的話if (t != null) {//定義個(gè)鎖final ReentrantLock mainLock = this.mainLock;//加鎖mainLock.lock();try {//獲取線程池狀態(tài)int rs = runStateOf(ctl.get());//做校驗(yàn)，不符合則拋IllegalThreadStateException異常if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();//否則把w對象加到workers中，workers是HashSet類型的對象 workers.add(w);int s = workers.size();//比較下，再賦個(gè)值if (s > largestPoolSize)largestPoolSize = s;//修改標(biāo)志，說明添加成功了workerAdded = true;}} finally {//解鎖mainLock.unlock();}//如果添加成功了if (workerAdded) {//就調(diào)用start方法執(zhí)行任務(wù)，這個(gè)start做的就是運(yùn)行任務(wù)的事了t.start();//修改標(biāo)志位workerStarted = true;}}} finally {//最后處理下那些，添加成功了，但是沒有開始執(zhí)行的任務(wù)if (! workerStarted)//這個(gè)方法主要作用就是把以上任務(wù)移除掉addWorkerFailed(w);}//返回運(yùn)行結(jié)果return workerStarted;}

看完這部分，其實(shí)還是沒發(fā)現(xiàn)線程池到底是怎么做到線程復(fù)用的，不過可以發(fā)現(xiàn)有個(gè)關(guān)鍵類，接下來就看看它的相關(guān)源碼：

Worker類的相關(guān)源碼?

//可以發(fā)現(xiàn)Worker實(shí)現(xiàn)了Runnable private final class Worker extends AbstractQueuedSynchronizerimplements Runnable{//這是Worker類的構(gòu)造方法 Worker(Runnable firstTask) {//設(shè)置一下狀態(tài)位state，和AQS有關(guān)setState(-1); // inhibit interrupts until runWorker//賦值this.firstTask = firstTask;//拿到線程工廠，并創(chuàng)建一個(gè)線程，然后很騷氣的把Worker對象做為參數(shù)去做線程的初始化this.thread = getThreadFactory().newThread(this);} }

實(shí)現(xiàn)了Runnable接口，那必然是要重寫run方法的，看看Worker類的run方法：?

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

看看runWorker方法源碼：?

final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;//設(shè)置為null，幫助gc回收w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {//這步做的通過取Worker對象的firstTask或者通過getTask方法從工作隊(duì)列中獲取待執(zhí)行的任務(wù)，只要不為null，就一直會循環(huán)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 {//然后直接調(diào)用task的run方法來執(zhí)行具體的任務(wù)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);}}

分析到這里，關(guān)鍵出現(xiàn)在task.run()上，首先我們知道task是Runnable類型的，直接調(diào)用run方法的話，JVM是不會幫我們?nèi)ド尚戮€程的，就像和調(diào)用普通方法一樣，所以一個(gè)線程始終都會在whlie循環(huán)的邏輯中不斷的被重復(fù)利用，然后去取Worker對象的firstTask或者通過getTask方法從工作隊(duì)列中獲取待執(zhí)行的任務(wù)，再直接調(diào)用Runnable類型的run方法執(zhí)行任務(wù)。

總結(jié)

簡單的從源碼來總結(jié)下，執(zhí)行任務(wù)先調(diào)用的是t.start()方法，這個(gè)t是個(gè)Thread對象，而這個(gè)Thread對象則是從Worker對象里獲得的，在Worker在做初始化時(shí)就會賦值Thread，同時(shí)Worker初始化Thread對象時(shí)又是以自己作為參數(shù)來完成，而Worker對象又是個(gè)實(shí)現(xiàn)了Runnable接口的類，那Worker對象就肯定有自己的run方法，所以t.start()方法真正意義上調(diào)用的是Worker對象中重寫的run方法，而這個(gè)Worker對象中的run方法里沒有所謂的本地start方法，JVM自然不會再創(chuàng)建新的線程，而是把它當(dāng)普通方法一樣執(zhí)行。再加上whlie循環(huán)體，這樣就做到了Worker對象新建的線程始終都會在一個(gè)大循環(huán)里，而這個(gè)線程會反復(fù)的獲取任務(wù)，接著執(zhí)行任務(wù)，知道任務(wù)都執(zhí)行完畢，這就是線程池實(shí)現(xiàn)“線程復(fù)用”的原理。

?

總結(jié)

以上是生活随笔為你收集整理的Executor家族的辨析的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。