ThreadPoolExecutor（線程池）源碼分析
阿新 • • 發佈：2017-12-24
情況 failed rac hashset 必須 ext 阻塞 throw ice
1. 常量和變量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 高3位為線程池的運行狀態，低29位為當前線程總數 private static final int COUNT_BITS = Integer.SIZE - 3; // 32 -3 = 29 private static final int CAPACITY = (1 << COUNT_BITS) - 1; // 線程池容量：2^29 - 1（0001,1111,1111,1111,1111,1111,1111,1111） privatestatic final int RUNNING = -1 << COUNT_BITS; // 1110,0000,0000,0000,0000,0000,0000,0000 private static final int SHUTDOWN = 0 << COUNT_BITS; // 0000,0000,0000,0000,0000,0000,0000,0000 private static final int STOP = 1 << COUNT_BITS; // 0010,0000,0000,0000,0000,0000,0000,0000 private static final int TIDYING = 2 << COUNT_BITS; // 0100,0000,0000,0000,0000,0000,0000,0000 private static finalint TERMINATED = 3 << COUNT_BITS; // 0110,0000,0000,0000,0000,0000,0000,0000 // RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED // 1. 調用shutdown方法：RUNNING -> SHUTDOWN -> 中斷workers中所有空閑的工作線程（getTask中） // 立刻調用tryTerminate方法：SHUTDOWN -> TIDYING -> TERMINATED，若workQueue不為空，則線程池運行狀態保持為SHUTDOWN // 執行完workQueue中最後一個工作任務的工作線程在processWorkerExit中將調用tryTerminate方法：SHUTDOWN -> TIDYING -> TERMINATED// 2. 調用shutdownNow方法：RUNNING || SHUTDOWN -> STOP -> 中斷workers中所有已經啟動的工作線程，獲取和清空workQueue中所有尚未執行的工作任務 // 立刻調用tryTerminate方法：STOP -> TIDYING -> TERMINATED private static int ctlOf(int rs, int wc) { return rs | wc; } private static int runStateOf(int c) { return c & ~CAPACITY; } // 取線程池的運行狀態（ctl高3位） private static int workerCountOf(int c) { return c & CAPACITY; } // 取當前線程總數（ctl低29位） private static boolean isRunning(int c) { return c < SHUTDOWN; } private static boolean runStateLessThan(int c, int s) { return c < s; } private static boolean runStateAtLeast(int c, int s) { return c >= s; } private boolean compareAndIncrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect + 1); } private boolean compareAndDecrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect - 1); } private void decrementWorkerCount() { do {} while (! compareAndDecrementWorkerCount(ctl.get())); } private final ReentrantLock mainLock = new ReentrantLock(); // workers鎖 private final Condition termination = mainLock.newCondition(); // 等待線程池的運行狀態成為TERMINATED（見awaitTermination方法） private final HashSet<Worker> workers = new HashSet<Worker>(); // 工作線程集合 private final BlockingQueue<Runnable> workQueue; // 工作任務隊列 private volatile int corePoolSize; // 最大核心線程數 private volatile int maximumPoolSize; // 最大線程數 private volatile long keepAliveTime; // 空閑核心線程的存活時間 private volatile ThreadFactory threadFactory; // 線程工廠 private volatile RejectedExecutionHandler handler; // 默認為defaultHandler private volatile boolean allowCoreThreadTimeOut; // 允許空閑核心線程超時退出 private int largestPoolSize; // 線程總數的最大值 private long completedTaskCount; // 線程池執行的工作任務總數 private static final RejectedExecutionHandler defaultHandler = new AbortPolicy(); // 拋異常 private static final RuntimePermission shutdownPerm = new RuntimePermission("modifyThread");
2. 父類AbstractExecutorService
不詳細講AbstractExecutorService，將會涉及到FutrueTask類，有空大家可以研究下。
public abstract class AbstractExecutorService implements ExecutorService { protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { return new FutureTask<T>(runnable, value); } protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { return new FutureTask<T>(callable); } public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); // FutureTask execute(ftask); return ftask; } public <T> Future<T> submit(Runnable task, T result) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task, result); // FutureTask execute(ftask); return ftask; } public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); // FutureTask execute(ftask); return ftask; } ... ... }
3. 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)) { // 線程池正在運行 -> 在workQueue中添加工作任務 // 在workQueue中添加工作任務後需要確保兩件事情 // 1. 線程池正在運行 // 2. 在允許空閑核心線程超時退出的情況下：當前線程總數 > 0 int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) // 線程池正在關閉 -> 在workQueue中移除工作任務 // 線程池成功關閉 reject(command); else if (workerCountOf(recheck) == 0) // 當前線程總數為0：corePoolSize == 0 || 所有核心工作線程超時（見getTask） addWorker(null, false); // 創建非核心工作線程（未綁定工作任務：將在workQueue中取工作任務執行） } else if (!addWorker(command, false)) // 線程池正在關閉 || 未能在workQueue添加工作任務：創建非核心工作線程 reject(command); } private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { // 線程池運行狀態發生變化將回到此處 int c = ctl.get(); /*記錄ctl*/ int rs = runStateOf(c); // 線程池正在關閉 && （線程池運行狀態 > SHUTDOWN || 工作任務不為空 || 工作任務隊列為空） // 線程池運行狀態 > SHUTDOWN：STOP，不添加新的工作線程 // SHUTDOWN && 工作任務不為空：不執行新的工作任務，不添加新的工作線程 // SHUTDOWN && 工作任務隊列為空：workQueue不存在待執行的工作任務，不添加新的工作線程 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { // CAS(ctl)失敗 && 線程池運行狀態未發生變化：回到此處 int wc = workerCountOf(c); // 當前線程總數 >= 線程池容量 || 當前線程數 > 核心線程數（最大線程數） if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; if (compareAndIncrementWorkerCount(c)) /*CAS設置ctl++*/ break retry; c = ctl.get(); if (runStateOf(c) != rs) // 線程池運行狀態發生變化 continue retry; } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 加workers鎖 try { int rs = runStateOf(ctl.get()); // 線程池正在運行 || （線程池運行狀態為SHUTDOWN && 工作任務為空 ） if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) throw new IllegalThreadStateException(); workers.add(w); // 添加工作線程到workers int s = workers.size(); if (s > largestPoolSize) // 跟蹤和記錄線程總數的最大值 largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); // 釋放workers鎖 } if (workerAdded) { t.start(); // 啟動工作線程 workerStarted = true; // 工作線程已啟動 } } } finally { if (! workerStarted) // 工作線程未啟動 addWorkerFailed(w); } return workerStarted; } private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 加workers鎖 try { if (w != null) workers.remove(w); // 在workers中移除工作線程 decrementWorkerCount(); // 當前工作線程總數-- tryTerminate(); // 嘗試終止線程池（w可能為shutdown後最後一個退出的工作線程，當前線程必須負責終止線程池） } finally { mainLock.unlock(); // 釋放workers鎖 } } public boolean remove(Runnable task) { boolean removed = workQueue.remove(task); // 在workQueue中移除工作任務 tryTerminate(); // 嘗試終止線程池（task可能為shutdown後workQueue中的最後一個工作任務，當前線程必須負責終止線程池） return removed; } final void reject(Runnable command) { handler.rejectedExecution(command, this); // 默認拋異常 }
4. ThreadPoolExecutor.Worker
不詳細講AbstractSynchronizer，有空大家可以研究下RenentranLock。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { final Thread thread; // 實際線程 Runnable firstTask; // 首個工作任務 volatile long completedTasks; // 執行的工作任務總數 Worker(Runnable firstTask) { setState(-1); // state = -1 ：加鎖（未啟動，不被中斷） this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } public void run() { runWorker(this); // 執行工作任務 } void interruptIfStarted() { // 中斷已啟動的工作線程 Thread t; if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try { t.interrupt(); } catch (SecurityException ignore) { } } } public void lock() { acquire(1); } public boolean tryLock() { return tryAcquire(1); } public void unlock() { release(1); } // 釋放鎖：state++（已啟動，可被中斷） ... ... }
5. runWorker
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // Worker.state++（已啟動，允許中斷） boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { // 執行綁定的工作任務 || 執行workQueue中的工作任務 w.lock(); // 工作線程加鎖 // 線程池運行狀態 >= STOP || （getTask時被中斷（清除中斷位） && 線程池運行狀態 >= STOP && 工作線程未再次被中斷） if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); // 中斷工作線程 try { beforeExecute(wt, task); // noop 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); // noop } } finally { task = null; w.completedTasks++; // 跟蹤和記錄w執行的工作任務總數 w.unlock(); // 工作線程釋放鎖 } } completedAbruptly = false; // 工作任務執行過程中未產生異常 } finally { // 1. while正常退出（getTask == null）：completedAbruptly為false // 2. while非正常退出（工作任務執行過程中產生異常，如被中斷）：completedAbruptly為true processWorkerExit(w, completedAbruptly); } } private Runnable getTask() { boolean timedOut = false; for (;;) { // CAS(ctl)失敗 || keepAliveTime內未取到工作任務 || 取工作任務時被中斷：回到此處 int c = ctl.get(); /*記錄ctl*/ int rs = runStateOf(c); // 線程池運行狀態 >= SHUTDOWN && （線程池運行狀態 >= STOP || workQueue為空） if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); // 當前線程工作總數-- return null; } int wc = workerCountOf(c); boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 核心線程允許超時 || 當前線程總數 > 最大核心線程數 if ((wc > maximumPoolSize || // 當前線程總數 > 最大線程數 (timed && timedOut)) && // （核心線程允許超時 && 線程已超時） || 當前線程總數 > 最大核心線程數 (wc > 1 || workQueue.isEmpty())) { // 當前線程總數 > 1 || workQueue為空 if (compareAndDecrementWorkerCount(c)) /*CAS設置ctl--*/ return null; continue; } try { // 取工作任務（keepAliveTime內 || 永久等待） Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; // 超時：keepAliveTime內未取到工作任務 } catch (InterruptedException retry) { // 取工作任務時被中斷 timedOut = false; // 取工作任務時未超時 } } } protected void beforeExecute(Thread t, Runnable r) { } protected void afterExecute(Runnable r, Throwable t) { } private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) // 工作任務執行過程中產生異常，如被中斷 decrementWorkerCount(); // CAS設置ctl--（區別compareAndDecrementWorkerCount） final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 加workers鎖 try { completedTaskCount += w.completedTasks; // 跟蹤和記錄線程池執行的工作任務總數 workers.remove(w); // 在workers中移除當前工作線程 } finally { mainLock.unlock(); // 釋放workers鎖 } tryTerminate(); // 嘗試終止線程池（可能是shutdown後最後一個即將退出的工作線程） int c = ctl.get(); if (runStateLessThan(c, STOP)) { // 線程池運行狀態 < STOP if (!completedAbruptly) { // 工作任務執行過程中未產生異常 int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) // 最後一個超時退出的工作線程 && workQueue不為空 min = 1; if (workerCountOf(c) >= min) // 當前線程總數 >= min return; } addWorker(null, false); // 補充一個工作線程 } }
6. shutdown和shutdownNow
public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 加workers鎖 try { checkShutdownAccess(); advanceRunState(SHUTDOWN); // 若當前線程運行狀態 < SHUTDOWN，則當前運行狀態置為SHUTDOWN interruptIdleWorkers(); // 中斷所有空閑的工作線程 onShutdown(); // noop } finally { mainLock.unlock(); // 釋放workers鎖 } tryTerminate(); // 嘗試終止線程池 } private void advanceRunState(int targetState) { for (;;) { int c = ctl.get(); /*記錄ctl*/ if (runStateAtLeast(c, targetState) || // 當前線程運行狀態 >= targetState ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c)))) /*CAS設置ctl = targetState*/ break; } } private void interruptIdleWorkers() { interruptIdleWorkers(false); // 中斷所有空閑的工作線程 } private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 加workers鎖 try { for (Worker w : workers) { Thread t = w.thread; if (!t.isInterrupted() && w.tryLock()) { // 工作線程未被中斷 && 工作線程嘗試加鎖（正在執行的工作線程已加鎖，見runWorker方法） try { t.interrupt(); // 中斷工作線程 } catch (SecurityException ignore) { } finally { w.unlock(); // 工作線程釋放鎖 } } if (onlyOne) // 是否只中斷一個？ break; } } finally { mainLock.unlock(); // 釋放workers鎖 } } void onShutdown() { } private static final boolean ONLY_ONE = true; final void tryTerminate() { for (;;) { // CAS(ctl)失敗將回到此處 int c = ctl.get(); /*記錄ctl*/ // 線程池正在運行 || 線程池運行狀態 >= TIDYING || （線程運行狀態為SHUTDOWN && workQueue不為空） // 線程池正在運行：當前線程不終止線程池 // 線程池運行狀態 >= TIDYING：線程池即將終止，當前線程不終止線程池 // 線程運行狀態為SHUTDOWN && workQueue不為空：正在等待workQueue為空，當前線程不終止線程池 if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return; if (workerCountOf(c) != 0) { // 中斷一個空閑的工作線程：可能之前未被shutdown方法中斷的工作線程已執行完畢，最後阻塞在workQueue上 interruptIdleWorkers(ONLY_ONE); return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 加workers鎖 try { if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { /*CAS設置ctl = TIDYING*/ try { terminated(); // noop } finally { ctl.set(ctlOf(TERMINATED, 0)); // ctl = TERMINATED termination.signalAll(); // 喚醒所有等待線程池結束的線程 } return; } } finally { mainLock.unlock(); // 釋放workers鎖 } } } protected void terminated() { } // 可以在調用shutdown方法後調用awaitTermination方法等待線程池的運行狀態成為TERMINATED，即等待最後一個執行線程退出 public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { long nanos = unit.toNanos(timeout); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (;;) { if (runStateAtLeast(ctl.get(), TERMINATED)) return true; if (nanos <= 0) return false; nanos = termination.awaitNanos(nanos); } } finally { mainLock.unlock(); } }
public List<Runnable> shutdownNow() { // 中斷所有正在執行的工作線程，清空workers中空閑的工作線程 List<Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 加workers鎖 try { checkShutdownAccess(); advanceRunState(STOP); // 若當前線程運行狀態 < STOP，則當前運行狀態置為STOP interruptWorkers(); // 中斷所有已經啟動的工作線程 tasks = drainQueue(); // 獲取和清空workQueue中所有尚未執行的工作任務 } finally { mainLock.unlock(); // 釋放鎖 } tryTerminate(); // 嘗試終止線程池 return tasks; // return 所有尚未執行的工作任務 } private void interruptWorkers() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); // 加workers鎖 try { for (Worker w : workers) w.interruptIfStarted(); // w已經啟動則中斷t } finally { mainLock.unlock(); // 加wrokers鎖 } } private List<Runnable> drainQueue() { // 獲取和清空workQueue中所有尚未執行的工作任務 BlockingQueue<Runnable> q = workQueue; ArrayList<Runnable> taskList = new ArrayList<Runnable>(); q.drainTo(taskList); if (!q.isEmpty()) { for (Runnable r : q.toArray(new Runnable[0])) { if (q.remove(r)) taskList.add(r); } } return taskList; }
ThreadPoolExecutor（線程池）源碼分析