文章目錄

• Pre
• execute源碼分析
• • addWorker()解讀
  • Worker解讀

---

Pre

[并發(fā)編程] - Executor框架#ThreadPoolExecutor源碼解讀02
說了一堆結(jié)論性的東西，作為開發(fā)人員著實是不過癮，那這里我們就來剖根問底來看下線程池是如何工作的。

---

execute源碼分析

ThreadPoolExecutor te = new ThreadPoolExecutor(5,10,500,TimeUnit.SECONDS,new ArrayBlockingQueue(5));for (int i = 0; i < 6; i++) {te.submit(()->{System.out.println("i m task ："+Thread.currentThread().getName());});}

使用ThreadPoolExecutor 自定義了一個線程池

參數(shù)對應(yīng)如下

int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue

調(diào)用了 AbstractExecutorService#submit

public Future<?> submit(Runnable task) {if (task == null) throw new NullPointerException();RunnableFuture<Void> ftask = newTaskFor(task, null);execute(ftask);return ftask;}

最核心的方法 execute ，由子類ThredPoolExecutor實現(xiàn)

public void execute(Runnable command) {if (command == null)throw new NullPointerException();/** clt記錄著runState和workerCount*/int c = ctl.get();/** workerCountOf方法取出低29位的值，表示當(dāng)前活動的線程數(shù)；* 如果當(dāng)前活動線程數(shù)小于corePoolSize，則新建一個線程放從入線程池中；* 并把任務(wù)添加到該線程中。*/if (workerCountOf(c) < corePoolSize) {/** addWorker中的第二個參數(shù)表示限制添加線程的數(shù)量是根據(jù)corePoolSize來判斷還是maximumPoolSize來判斷；* 如果為true，根據(jù)corePoolSize來判斷；* 如果為false，則根據(jù)maximumPoolSize來判斷*/if (addWorker(command, true))return;/** 如果添加失敗，則重新獲取ctl值*/c = ctl.get();}/** 如果當(dāng)前線程池是運行狀態(tài)并且任務(wù)添加到隊列成功*/if (isRunning(c) && workQueue.offer(command)) {// 重新獲取ctl值int recheck = ctl.get();// 再次判斷線程池的運行狀態(tài)，如果不是運行狀態(tài)，由于之前已經(jīng)把command添加到workQueue中了，// 這時需要移除該command// 執(zhí)行過后通過handler使用拒絕策略對該任務(wù)進(jìn)行處理，整個方法返回if (! isRunning(recheck) && remove(command))reject(command);/** 獲取線程池中的有效線程數(shù)，如果數(shù)量是0，則執(zhí)行addWorker方法* 這里傳入的參數(shù)表示：* 1. 第一個參數(shù)為null，表示在線程池中創(chuàng)建一個線程，但不去啟動；* 2. 第二個參數(shù)為false，將線程池的有限線程數(shù)量的上限設(shè)置為maximumPoolSize，添加線程時根據(jù)maximumPoolSize來判斷；* 如果判斷workerCount大于0，則直接返回，在workQueue中新增的command會在將來的某個時刻被執(zhí)行。*/else if (workerCountOf(recheck) == 0)addWorker(null, false);}/** 如果執(zhí)行到這里，有兩種情況：* 1. 線程池已經(jīng)不是RUNNING狀態(tài)；* 2. 線程池是RUNNING狀態(tài)，但workerCount >= corePoolSize并且workQueue已滿。* 這時，再次調(diào)用addWorker方法，但第二個參數(shù)傳入為false，將線程池的有限線程數(shù)量的上限設(shè)置為maximumPoolSize；* 如果失敗則拒絕該任務(wù)*/else if (!addWorker(command, false))reject(command); }

主要的流程，注釋中也寫的很清楚了

/** Proceed in 3 steps:** 1. If fewer than corePoolSize threads are running, try to* start a new thread with the given command as its first* task. The call to addWorker atomically checks runState and* workerCount, and so prevents false alarms that would add* threads when it shouldn't, by returning false.** 2. If a task can be successfully queued, then we still need* to double-check whether we should have added a thread* (because existing ones died since last checking) or that* the pool shut down since entry into this method. So we* recheck state and if necessary roll back the enqueuing if* stopped, or start a new thread if there are none.** 3. If we cannot queue task, then we try to add a new* thread. If it fails, we know we are shut down or saturated* and so reject the task.*/

簡單來說，在執(zhí)行execute()方法時如果狀態(tài)一直是RUNNING時，的執(zhí)行過程如下：

如果workerCount < corePoolSize，則創(chuàng)建并啟動一個線程來執(zhí)行新提交的任務(wù)；

如果workerCount >= corePoolSize，且線程池內(nèi)的阻塞隊列未滿，則將任務(wù)添加到該阻塞隊列中；

如果workerCount >= corePoolSize && workerCount < maximumPoolSize，且線程池內(nèi)的阻塞隊列已滿，則創(chuàng)建并啟動一個線程來執(zhí)行新提交的任務(wù)；

如果workerCount >= maximumPoolSize，并且線程池內(nèi)的阻塞隊列已滿, 則根據(jù)拒絕策略來處理該任務(wù), 默認(rèn)的處理方式是直接拋異常。

Note : 這里要注意一下addWorker(null, false);，也就是創(chuàng)建一個線程，但并沒有傳入任務(wù)，因為任務(wù)已經(jīng)被添加到workQueue中了，所以worker在執(zhí)行的時候，會直接從workQueue中獲取任務(wù)。所以，在workerCountOf(recheck) == 0時執(zhí)行addWorker(null, false);也是為了保證線程池在RUNNING狀態(tài)下必須要有一個線程來執(zhí)行任務(wù)。

---

addWorker()解讀

private boolean addWorker(Runnable firstTask, boolean core) {}

addWorker方法的主要工作是在線程池中創(chuàng)建一個新的線程并執(zhí)行，

• firstTask參數(shù) 用于指定新增的線程執(zhí)行的第一個任務(wù)，
• core參數(shù)為true表示在新增線程時會判斷當(dāng)前活動線程數(shù)是否少于corePoolSize，false表示新增線程前需要判斷當(dāng)前活動線程數(shù)是否少于maximumPoolSize

private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();// 獲取運行狀態(tài)int rs = runStateOf(c);/** 這個if判斷* 如果rs >= SHUTDOWN，則表示此時不再接收新任務(wù)；* 接著判斷以下3個條件，只要有1個不滿足，則返回false：* 1. rs == SHUTDOWN，這時表示關(guān)閉狀態(tài)，不再接受新提交的任務(wù)，但卻可以繼續(xù)處理阻塞隊列中已保存的任務(wù)* 2. firsTask為空* 3. 阻塞隊列不為空* * 首先考慮rs == SHUTDOWN的情況* 這種情況下不會接受新提交的任務(wù)，所以在firstTask不為空的時候會返回false；* 然后，如果firstTask為空，并且workQueue也為空，則返回false，* 因為隊列中已經(jīng)沒有任務(wù)了，不需要再添加線程了*/// Check if queue empty only if necessary.if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {// 獲取線程數(shù)int wc = workerCountOf(c);// 如果wc超過CAPACITY，也就是ctl的低29位的最大值（二進(jìn)制是29個1），返回false；// 這里的core是addWorker方法的第二個參數(shù)，如果為true表示根據(jù)corePoolSize來比較，// 如果為false則根據(jù)maximumPoolSize來比較。// if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;// 嘗試增加workerCount，如果成功，則跳出第一個for循環(huán)if (compareAndIncrementWorkerCount(c))break retry;// 如果增加workerCount失敗，則重新獲取ctl的值c = ctl.get(); // Re-read ctl// 如果當(dāng)前的運行狀態(tài)不等于rs，說明狀態(tài)已被改變，返回第一個for循環(huán)繼續(xù)執(zhí)行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 {// 根據(jù)firstTask來創(chuàng)建Worker對象w = new Worker(firstTask);// 每一個Worker對象都會創(chuàng)建一個線程final Thread t = w.thread;if (t != null) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {int rs = runStateOf(ctl.get());// rs < SHUTDOWN表示是RUNNING狀態(tài)；// 如果rs是RUNNING狀態(tài)或者rs是SHUTDOWN狀態(tài)并且firstTask為null，向線程池中添加線程。// 因為在SHUTDOWN時不會在添加新的任務(wù)，但還是會執(zhí)行workQueue中的任務(wù)if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();// workers是一個HashSetworkers.add(w);int s = workers.size();// largestPoolSize記錄著線程池中出現(xiàn)過的最大線程數(shù)量if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {// 啟動線程t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted; }

---

Worker解讀

addWorker中多次提到了這個Work這個類， 其實就是 線程池中的每一個線程被封裝成一個Worker對象，ThreadPool維護(hù)的其實就是一組Worker對象

ThreadPoolExector中內(nèi)部類 Worker

private final class Worker extends AbstractQueuedSynchronizer implements Runnable{private static final long serialVersionUID = 6138294804551838833L;final Thread thread;Runnable firstTask;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 runWorkerthis.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) {}}}}

繼承了AQS，并實現(xiàn)了Runnable接口 ，
兩個比較重要的屬性

firstTask用它來保存?zhèn)魅氲娜蝿?wù)；

thread是在調(diào)用構(gòu)造方法時通過ThreadFactory來創(chuàng)建的線程，是用來處理任務(wù)的線程。

在調(diào)用構(gòu)造方法時，需要把任務(wù)傳入，這里通過getThreadFactory().newThread(this);來新建一個線程，newThread方法傳入的參數(shù)是this，因為Worker本身繼承了Runnable接口，也就是一個線程，所以一個Worker對象在啟動的時候會調(diào)用Worker類中的run方法。

Worker繼承了AQS，使用AQS來實現(xiàn)獨占鎖的功能。為什么不使用ReentrantLock來實現(xiàn)呢？可以看到tryAcquire方法，它是不允許重入的，而ReentrantLock是允許重入的

總結(jié)

以上是生活随笔為你收集整理的[并发编程] - Executor框架#ThreadPoolExecutor源码解读03的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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