內容目錄

• TreadPoolexecutor原始碼解析
• 類關係圖
• Executor介面
• ExecutorService介面
• AbstractExecutorService
• 成員變數
• 建構函式
• Worker類解析
• Worker簡介
• 成員變數
• 構造方法
• execute()解析
• addWorker()解析
• runWorker()解析
• getTask()解析
• processWorkerExit()解析
• tryTerminate()解析
• interruptIdleWorker()解析
• 監控執行緒池

TreadPoolexecutor原始碼解析

1、類關係圖

2、Executor介面

Executor提供execute()用來啟動任務

 public interface Executor {
 //用來啟動任務
 void execute(Runnable command);
 }

3、ExecutorService介面

ExecutorService提供了一些管理執行緒池方法和任務執行的方法

 public interface ExecutorService extends Executor {
 //關閉執行緒池，佇列已經存在的任務可以繼續執行
 void shutdown();
 //關閉執行緒池，中斷未執行的任務
 List<Runnable> shutdownNow();
 //判斷是否關閉
 boolean isShutdown();
 //判斷是否終止
 boolean isTerminated();
 //設定超時終止
 boolean awaitTermination(long timeout, TimeUnit unit)
 throws InterruptedException;
 //提交Callable任務
 <T> Future<T> submit(Callable<T> task);
 //提交Runable任務，帶返回值
 <T> Future<T> submit(Runnable task, T result);
 //提交Runnable任務不帶返回值
 Future<?> submit(Runnable task);
 //invokeAll()是同步的，其需要等待任務的完成，才能返回。submit()是非同步的
 <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
 throws InterruptedException;
 <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
 long timeout, TimeUnit unit)
 throws InterruptedException;
 //invokeAny()取第一個任務的返回值，然後呼叫interrupt方法中斷其它任務。
 <T> T invokeAny(Collection<? extends Callable<T>> tasks)
 throws InterruptedException, ExecutionException;
 <T> T invokeAny(Collection<? extends Callable<T>> tasks,
 long timeout, TimeUnit unit)
 throws InterruptedException, ExecutionException, TimeoutException;
 }

4、AbstractExecutorService

AbstractExecutorService類重寫某些方法。自定義了newTaskFor()用於構建RunnableFuture。

5、成員變數

/**
 * 執行緒池使用一個int變數儲存執行緒池狀態和工作執行緒數
 * int4個位元組，32位，用高三位儲存執行緒池狀態，低29位儲存工作執行緒數
 * 為什麼使用一個變數來同時表示執行緒狀態和執行緒數？就是節省空間。諮詢了一下寫c的朋友，他們經常這麼寫
 **/
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//COUNT_BITS=29
private static final int COUNT_BITS = Integer.SIZE - 3;
//理論上執行緒池最大執行緒數量CAPACITY=536870911
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
/**
 * 執行緒池狀態轉換
 * RUNNING -> SHUTDOWN
 * RUNNING or SHUTDOWN -> STOP
 * SHUTDOWN or STOP -> TIDYING
 * TIDYING -> TERMINATED terminated()執行完後變為該TERMINATED
 */
//接受新任務，可以處理阻塞佇列裡的任務
private static final int RUNNING = -1 << COUNT_BITS;
//不接受新任務，可以處理阻塞佇列裡的任務。執行shutdown()會變為SHUTDOWN
private static final int SHUTDOWN = 0 << COUNT_BITS;
//不接受新的任務，不處理阻塞佇列裡的任務，中斷正在處理的任務。執行shutdownNow()會變為STOP
private static final int STOP = 1 << COUNT_BITS;
//臨時過渡狀態，所有的任務都執行完了，當前執行緒池有效的執行緒數量為0，這個時候執行緒池的狀態是TIDYING，執行terminated()變為TERMINATED
private static final int TIDYING = 2 << COUNT_BITS;
//終止狀態，terminated()呼叫完成後的狀態
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; }
//初始化ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }
//用於儲存等待執行的任務的阻塞佇列。比如LinkedBlockQueue，SynchronousQueue等
private final BlockingQueue<Runnable> workQueue;
//重入鎖，更新執行緒池核心大小、執行緒池最大大小等都有用到
private final ReentrantLock mainLock = new ReentrantLock();
//用於儲存woker
private final HashSet<Worker> workers = new HashSet<Worker>();
//用於終止執行緒池
private final Condition termination = mainLock.newCondition();
//記錄執行緒池中曾經出現過的最大執行緒數
private int largestPoolSize;
//完成任務數量
private long completedTaskCount; 
//執行緒工廠
private volatile ThreadFactory threadFactory;
/**
 * rejectedExecutionHandler：任務拒絕策略
 * DiscardOldestPolicy：丟棄佇列裡最近的一個任務，並執行當前任務
 * AbortPolicy：丟擲異常。這也是預設的策略
 * CallerRunsPolicy：用呼叫者所線上程來執行任務
 * DiscardPolicy：不處理，丟棄掉
 */
private volatile RejectedExecutionHandler handler;
/**
 * 執行緒空閒時間
 * 當執行緒空閒時間達到keepAliveTime，該執行緒會退出，直到執行緒數量等於corePoolSize。
 * 如果allowCoreThreadTimeout設定為true，則所有執行緒均會退出。
 */
private volatile long keepAliveTime;
//是否允許核心執行緒空閒超時退出，預設值為false。
private volatile boolean allowCoreThreadTimeOut;
/**
 * 核心執行緒數
 * 核心執行緒會一直存活，即使沒有任務需要處理，當執行緒數小於核心執行緒數時。
 * 即使現有的執行緒空閒，執行緒池也會優先建立新執行緒來處理任務，而不是直接交給現有的執行緒處理。
 * 核心執行緒數在初始化時不會建立，只有提交任務的時候才會建立。核心執行緒在allowCoreThreadTimeout為true的時候超時會退出。
 */
private volatile int corePoolSize;
 /** 最大執行緒數
 * 當執行緒數大於或者等於核心執行緒，且任務佇列已滿時，執行緒池會建立新的執行緒，直到執行緒數量達到maxPoolSize。
 * 如果執行緒數已等於maxPoolSize，且任務佇列已滿，則已超出執行緒池的處理能力，執行緒池會採取拒絕操作。
 */
private volatile int maximumPoolSize;
//預設的拒絕策略：丟擲異常
private static final RejectedExecutionHandler defaultHandler =
 new AbortPolicy();
private static final RuntimePermission shutdownPerm =
 new RuntimePermission("modifyThread");

6、建構函式

//直接提供了一個最終呼叫的建構函式
//大致邏輯就是給執行緒池核心引數賦值
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;
}

Worker類解析

1、Worker簡介

Woeker類是ThreadPoolExecutor一個內部類。此類繼AbstractQueuedSynchronizer，

目的在於判斷執行緒是否空閒以及是否可以被中斷。實現Runnable，在run()中呼叫了runWorker()。

Worker類中firstTask用來儲存傳入的任務，thread是在呼叫構造方法時通過ThreadFactory來建立的執行緒，是用來處理任務的執行緒。注意：這個執行緒並不是task執行緒。

2、成員變數

 /** 工作執行緒空間，由執行緒池中所設定的執行緒工廠建立*/
 final Thread thread;
 /** 在構造方法中傳入的任務*/
 Runnable firstTask;
 /** 執行完任務的總數*/
 volatile long completedTasks;

3、構造方法

 Worker(Runnable firstTask) {
 //state設定為-1是為了禁止在執行任務前對任務進行中斷。
 setState(-1); 
 //提交的task
 this.firstTask = firstTask;
 //從執行緒工廠獲取的執行緒，注意這個thread並不是使用者執行緒
 this.thread = getThreadFactory().newThread(this);
 }

4、execute()解析

public void execute(Runnable command) {
 //判斷提交的任務是否為空
 if (command == null)
 throw new NullPointerException();
 //獲取執行緒池狀態和工作執行緒數量結合體(下文統稱為ctl)
 int c = ctl.get();
 //判斷工作執行緒數量是否小於核心執行緒數
 if (workerCountOf(c) < corePoolSize) {
 //把任務新增到worker，新增成功則返回
 if (addWorker(command, true))
 return;
 //再次獲取ctl
 c = ctl.get();
 }
 //如果執行緒池狀態是正在執行並且新增一個任務到佇列成功
 if (isRunning(c) && workQueue.offer(command)) {
 int recheck = ctl.get();
 //再次校驗執行緒池狀態，如果狀態不是RUNNING則需要從佇列中移除任務
 if (! isRunning(recheck) && remove(command))
 //執行拒絕策略
 reject(command);
 else if (workerCountOf(recheck) == 0)
 //進入這裡說明空閒核心執行緒數都超時退出啦
 //因為任務已經放入隊列了，所以此處不需要傳入任務
 //注意事項，網上很多說這裡建立一個執行緒不啟動這是錯誤的。博主親測這裡建立了一個執行緒並且需要啟動
 addWorker(null, false);
 }
 /*
 * 如果執行到這裡有兩種情況：
 * 執行緒池已經不是RUNNING狀態
 * 執行緒池是RUNNING狀態，workerCount>=corePoolSize並且workQueue已滿
 */
 //呼叫addWorker(),傳入false代表把執行緒池執行緒數量設定maximumPoolSize，如果新增失敗則執行拒絕策略。
 else if (!addWorker(command, false))
 reject(command);
}

5、addWorker()解析

/**
 *@param firstTask 表示執行的任務；
 *@param core 表示限制新增執行緒的數量是根據corePoolSize來判斷還是maximumPoolSize來判斷； 
 * 如果為true，根據corePoolSize來判斷；如果為false，則根據maximumPoolSize來判斷
 */
private boolean addWorker(Runnable firstTask, boolean core) {
 retry:
 for (;;) {
 int c = ctl.get();
 //獲取執行緒池狀態
 int rs = runStateOf(c);
 /**
 * 只要滿足下面任一條就直接返回false
 * 執行緒池狀態為STOP, TYDING 或 TERMINATD 狀態
 * 執行緒池狀態為SHUTDOWN，並且firstTask != null 或者workQueue為空
 */
 //這裡為什麼不直接判斷執行緒池狀態呢？是因為有可能線上程池狀態為RUNNING時已經把任務放入佇列中，放入完成以後狀態變為SHUTDOWN
 if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&
 firstTask == null &&! workQueue.isEmpty()))
 return false;
 for (;;) {
 //得到工作的執行緒數量
 int wc = workerCountOf(c);
 //如果工作執行緒數量大於理論上執行緒池容量；或者工作執行緒數大於(corePoolSize or maximumPoolSize) 直接返回false，新增失敗
 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
 return false;
 //cas修改工作執行緒數，工作執行緒數+1。如果修改失敗需要重新執行；成功退出迴圈
 if (compareAndIncrementWorkerCount(c))
 break retry;
 c = ctl.get(); 
 //如果執行緒池狀態變化則需要重新執行
 if (runStateOf(c) != rs)
 continue retry;
 }
 }
 //worker是否已經啟動
 boolean workerStarted = false;
 //worker是否新增成功
 boolean workerAdded = false;
 Worker w = null;
 try {
 //構建worker
 w = new Worker(firstTask);
 //注意，這個thread不是firstTask，是從執行緒工廠造出來的
 final Thread t = w.thread;
 if (t != null) {
 //獲取鎖
 final ReentrantLock mainLock = this.mainLock;
 mainLock.lock();
 try {
 //獲取執行緒池狀態
 int rs = runStateOf(ctl.get());
 //如果執行緒池狀態是RUNNING或者狀態是SHUTDOWN但是佇列裡面還有任務
 if (rs < SHUTDOWN ||
 (rs == SHUTDOWN && firstTask == null)) {
 //如果t.isAlive()=true，說明是有問題的，都沒有啟動，t怎麼會是活的呢。所以丟擲異常。
 if (t.isAlive()) 
 throw new IllegalThreadStateException();
 //把worker新增到set集合
 workers.add(w);
 //記錄執行緒池中出現的最大執行緒數
 int s = workers.size();
 if (s > largestPoolSize)
 largestPoolSize = s;
 //新增成功標識
 workerAdded = true;
 }
 } finally {
 //釋放鎖
 mainLock.unlock();
 }
 //新增成功啟動執行緒，啟動執行緒是呼叫了runWorker()
 if (workerAdded) {
 t.start();
 workerStarted = true;
 }
 }
 } finally {
 //啟動失敗
 if (! workerStarted)
 //啟動失敗需要從workers中移除當前構造的woker；工作執行緒數減1了；執行tryTerminate()判斷是否終止執行緒池。
 addWorkerFailed(w);
 }
 return workerStarted;
}

6、runWorker()解析

final void runWorker(Worker w) {
 //獲取當前執行緒
 Thread wt = Thread.currentThread();
 //需要執行的任務
 Runnable task = w.firstTask;
 //置為null
 w.firstTask = null;
 //這裡是為了把之前的state=-1設定為state=0，此時允許中斷
 w.unlock(); 
 //是否異常退出迴圈
 boolean completedAbruptly = true;
 try {
 //如果任務不為null或者getTask()不為null
 while (task != null || (task = getTask()) != null) {
 //獲取鎖。這裡使用鎖的目的在於標識正在處理任務
 w.lock();
 //執行緒池=SHUWDOWN，要保證當前執行緒是中斷狀態
 //執行緒池!=SHUWDOWN，要保證當前執行緒不是中斷狀態
 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 {
 //一定要注意。執行到這裡說明getTask()返回null。說明當前執行緒池中不需要那麼多執行緒來執行任務了，可以把多於corePoolSize數量的工作執行緒幹掉
 processWorkerExit(w, completedAbruptly);
 }
}

7、getTask()解析

//什麼情況才會執行getTask呢？說明工作執行緒數已經大於核心執行緒數才會執行getTask()。一定要記住這一點
private Runnable getTask() {
 //表示上次從阻塞佇列中取任務時是否超時
 boolean timedOut = false; 
 for (;;) {
 int c = ctl.get();
 //獲取執行緒池狀態
 int rs = runStateOf(c);
 //如果執行緒池狀態rs >= SHUTDOWN並且(rs >= STOP，或者阻塞佇列為空)。則將workerCount減1並返回null。
 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
 decrementWorkerCount();
 return null;
 }
 //獲取工作執行緒數
 int wc = workerCountOf(c);
 //表示是否需要超時控制。allowCoreThreadTimeOut預設false；如果執行緒池數量超過核心執行緒數也是需要超時控制的
 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
 //校驗工作執行緒數量和任務佇列是否為空
 if ((wc > maximumPoolSize || (timed && timedOut))
 && (wc > 1 || workQueue.isEmpty())) {
 //工作執行緒數量-1
 if (compareAndDecrementWorkerCount(c))
 return null;
 continue;
 }
 try {
 //根據是否超時從佇列中獲取任務
 Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
 if (r != null)
 return r;
 timedOut = true;
 } catch (InterruptedException retry) {
 //說明發生了中斷
 timedOut = false;
 }
 }
}

8、processWorkerExit()解析

private void processWorkerExit(Worker w, boolean completedAbruptly) {
 //是否異常，如果異常工作執行緒數量-1
 if (completedAbruptly) 
 decrementWorkerCount();
 final ReentrantLock mainLock = this.mainLock;
 mainLock.lock();
 try { 
 //增加執行緒池完成任務數量
 completedTaskCount += w.completedTasks;
 //從worker中移除完成任務
 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; // replacement not needed
 }
 addWorker(null, false);
 }
}

9、tryTerminate()解析

 //根據執行緒池狀態判斷是否結束執行緒池
final void tryTerminate() {
 for (;;) {
 int c = ctl.get();
 //RUNNING不能結束執行緒池
 //執行緒池狀態是TIDYING或TERMINATED說明執行緒池已經處於正在終止的路上
 //狀態為SHUTDOWN，但是任務佇列不為空不能結束執行緒池
 if (isRunning(c) ||
 runStateAtLeast(c, TIDYING) ||
 (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
 return;
 //工作執行緒數量不等於0，中斷一個空閒的工作執行緒，並返回
 if (workerCountOf(c) != 0) { 
 interruptIdleWorkers(ONLY_ONE);
 return;
 }
 final ReentrantLock mainLock = this.mainLock;
 mainLock.lock();
 try {
 // 設定執行緒池狀態為TIDYING，如果設定成功，則呼叫terminated方法
 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
 try { 
 //子類實現
 terminated();
 } finally {
 // 設定狀態為TERMINATED
 ctl.set(ctlOf(TERMINATED, 0));
 termination.signalAll();
 }
 return;
 }
 } finally {
 mainLock.unlock();
 }
 }
}

10、interruptIdleWorker()解析

//該方法用於給空閒工作執行緒一箇中斷標識
private void interruptIdleWorkers(boolean onlyOne) {
 final ReentrantLock mainLock = this.mainLock;
 mainLock.lock();
 try {
 //遍歷worker，根據onlyOne判斷，如果為ture只中斷一個空閒執行緒
 for (Worker w : workers) {
 Thread t = w.thread;
 //執行緒沒有被中斷並且執行緒是空閒狀態tryLock()判斷是否空閒
 if (!t.isInterrupted() && w.tryLock()) {
 try {
 t.interrupt();
 } catch (SecurityException ignore) {
 } finally {
 w.unlock();
 }
 }
 if (onlyOne)
 break;
 }
 } finally {
 mainLock.unlock();
 }
}

監控執行緒池

通過執行緒池提供的引數進行監控

• getTaskCount：執行緒池任務總數。
• getCompletedTaskCount：執行緒池已完成的任務數量，小於等於completedTaskCount。
• getPoolSize：執行緒池當前的核心執行緒數量。
• getLargestPoolSize：執行緒池曾經建立過的最大執行緒數量。
• getActiveCount：當前執行緒池中正在執行任務的執行緒數量。