java.util.concurrent.ThreadPoolExecutor 原始碼
阿新 • • 發佈：2018-11-29
執行緒池相關
原始碼：
package java.util.concurrent; import java.security.AccessControlContext; import java.security.AccessController; import java.security.PrivilegedAction; import java.util.ArrayList; import java.util.ConcurrentModificationException; import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.AbstractQueuedSynchronizer; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; public class ThreadPoolExecutor extends AbstractExecutorService { // 執行緒池的控制狀態（用來表示執行緒池的執行狀態（整形的高3位）和執行的worker數量（低29位）） private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 29位的偏移量 private static final int COUNT_BITS = Integer.SIZE - 3; // 最大容量（2^29 - 1） private static final int CAPACITY = (1 << COUNT_BITS) - 1; // 執行緒執行狀態，總共有5個狀態，需要3位來表示（所以偏移量的29 = 32 - 3） /** * RUNNING：接受新任務並且處理已經進入阻塞佇列的任務 * SHUTDOWN：不接受新任務，但是處理已經進入阻塞佇列的任務 * STOP：不接受新任務，不處理已經進入阻塞佇列的任務並且中斷正在執行的任務 * TIDYING：所有的任務都已經終止，workerCount為0， 執行緒轉化為TIDYING狀態並且呼叫terminated鉤子函式 * TERMINATED：terminated鉤子函式已經執行完成 **/ private static final int RUNNING = -1 << COUNT_BITS; private static final int SHUTDOWN = 0 << COUNT_BITS; private static final int STOP = 1 << COUNT_BITS; private static final int TIDYING = 2 << COUNT_BITS; private static final int TERMINATED = 3 << COUNT_BITS; // 阻塞佇列 private final BlockingQueue<Runnable> workQueue; // 可重入鎖 private final ReentrantLock mainLock = new ReentrantLock(); // 存放工作執行緒集合 private final HashSet<Worker> workers = new HashSet<Worker>(); // 終止條件 private final Condition termination = mainLock.newCondition(); // 最大執行緒池容量 private int largestPoolSize; // 已完成任務數量 private long completedTaskCount; // 執行緒工廠 private volatile ThreadFactory threadFactory; // 拒絕執行處理器 private volatile RejectedExecutionHandler handler; // 執行緒等待執行時間 private volatile long keepAliveTime; // 是否執行核心執行緒超時 private volatile boolean allowCoreThreadTimeOut; // 核心池的大小 private volatile int corePoolSize; // 最大執行緒池大小 private volatile int maximumPoolSize; // 預設拒絕執行處理器 private static final RejectedExecutionHandler defaultHandler = new AbortPolicy(); private static final RuntimePermission shutdownPerm = new RuntimePermission("modifyThread"); private final AccessControlContext acc; //核心內部類 private final class Worker extends AbstractQueuedSynchronizer implements Runnable { private static final long serialVersionUID = 6138294804551838833L;// 版本號 final Thread thread;// worker 所對應的執行緒 Runnable firstTask;// worker所對應的第一個任務 volatile long completedTasks;// 已完成任務數量 //建構函式 Worker(Runnable firstTask) { setState(-1);// 設定AQS的state為-1 this.firstTask = firstTask;// 初始化第一個任務 this.thread = getThreadFactory().newThread(this);// 根據當前worker，初始化執行緒 } // 重寫了Runnable的run方法 public void run() { runWorker(this); } // 是否被獨佔,0代表未被獨佔，1代表被獨佔 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);// 設定獨佔執行緒為null setState(0);// 設定狀態為0 return true; } // 獲取鎖 public void lock() { acquire(1); } // 嘗試獲取鎖 public boolean tryLock() { return tryAcquire(1); } // 釋放鎖 public void unlock() { release(1); } // 是否被獨佔 public boolean isLocked() { return isHeldExclusively(); } //如果worker對應的執行緒已經啟動,則中斷該執行緒的執行 void interruptIfStarted() { Thread t; if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {// AQS狀態大於等於0並且worker對應的執行緒不為null並且該執行緒沒有被中斷 try { t.interrupt();// 中斷執行緒 } catch (SecurityException ignore) { } } } } //拒絕策略繼承介面RejectedExecutionHandler，實現了rejectedExecution方法： //rejectedExecution：當execute方法不能接受某個任務時，可以由 ThreadPoolExecutor 呼叫的方法。 //因為超出其界限而沒有更多可用的執行緒或佇列槽時，或者關閉 Executor 時就可能發生這種情況。 //在沒有其他替代方法的情況下，該方法可能丟擲未經檢查的 RejectedExecutionException，而該異常將傳播到 execute 的呼叫者。 public static class CallerRunsPolicy implements RejectedExecutionHandler { //構造器 public CallerRunsPolicy() { } //執行呼叫者執行緒中的任務r；如果執行程式已關閉，則會丟棄該任務 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) {//執行程式未關閉 r.run();//執行任務r } } } //拒絕策略繼承介面RejectedExecutionHandler，實現了rejectedExecution方法 public static class AbortPolicy implements RejectedExecutionHandler { //構造器 public AbortPolicy() { } //用於被拒絕任務的處理程式，它總是丟擲 RejectedExecutionException public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString()); } } //拒絕策略繼承介面RejectedExecutionHandler，實現了rejectedExecution方法 public static class DiscardPolicy implements RejectedExecutionHandler { //構造器 public DiscardPolicy() { } //用於被拒絕任務的處理程式，不執行任何操作：將丟棄被拒絕的任務 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { } } //拒絕策略繼承介面RejectedExecutionHandler，實現了rejectedExecution方法 public static class DiscardOldestPolicy implements RejectedExecutionHandler { //構造器 public DiscardOldestPolicy() { } //用於被拒絕任務的處理程式，它放棄最舊的未處理請求，然後重試execute；如果執行程式已關閉，則會丟棄該任務。 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) {//執行程式未關閉 e.getQueue().poll();//放棄最舊的未處理請求 e.execute(r);//重試r } } } //構造器 public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); } //構造器 public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,threadFactory, defaultHandler); } //構造器 public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), handler); } //構造器 public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) { if (corePoolSize < 0 ||// 核心大小不能小於0 maximumPoolSize <= 0 ||// 執行緒池的初始最大容量不能小於0 maximumPoolSize < corePoolSize ||// 初始最大容量不能小於核心大小 keepAliveTime < 0)// keepAliveTime不能小於0 throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null)//檢查引數合法性 throw new NullPointerException(); // 初始化相應的域 this.acc = System.getSecurityManager() == null ? null : AccessController.getContext(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; } //當在客戶端呼叫submit()時，之後會間接呼叫到execute函式，其在將來某個時間執行給定任務，此方法中並不會直接執行給定的任務 public void execute(Runnable command) { if (command == null)// 命令為null，丟擲異常 throw new NullPointerException(); /* * 進行下面三步： * 1. 如果執行的執行緒小於corePoolSize,則嘗試使用使用者定義的Runnalbe物件建立一個新的執行緒 * 呼叫addWorker函式會原子性的檢查runState和workCount，通過返回false來防止在不應該新增執行緒時添加了執行緒 * * 2. 如果一個任務能夠成功入佇列，在新增一個線城時仍需要進行雙重檢查（因為在前一次檢查後可能該執行緒死亡了）， * 或者當進入到此方法時，執行緒池已經shutdown了，所以需要再次檢查狀態，若有必要，當停止時還需要回滾入佇列操作， * 或者當執行緒池沒有執行緒時需要建立一個新執行緒 * * 3. 如果無法入佇列，那麼需要增加一個新執行緒， * 如果此操作失敗，那麼就意味著執行緒池已經shutdown或者已經飽和了，所以拒絕任務 */ int c = ctl.get();// 獲取執行緒池控制狀態 if (workerCountOf(c) < corePoolSize) {//執行的執行緒小於corePoolSize if (addWorker(command, true))//嘗試使用使用者定義的Runnalbe物件建立一個新的執行緒 return; c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (!isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false))//如果無法入佇列，那麼需要增加一個新執行緒 reject(command);//操作失敗，意味著執行緒池已經shutdown或者已經飽和,所以拒絕任務 } //按過去執行已提交任務的順序發起一個有序的關閉，但是不接受新任務。 //首先會檢查是否具有shutdown的許可權，然後設定執行緒池的控制狀態為SHUTDOWN，之後中斷空閒的worker，最後嘗試終止執行緒池 public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 檢查shutdown許可權 checkShutdownAccess(); // 設定執行緒池控制狀態為SHUTDOWN advanceRunState(SHUTDOWN); // 中斷空閒worker interruptIdleWorkers(); // 呼叫shutdown鉤子函式 onShutdown(); } finally { mainLock.unlock(); } // 嘗試終止 tryTerminate(); } //嘗試停止所有的活動執行任務、暫停等待任務的處理，並返回等待執行的任務列表。 //會終止所有的worker，而並非只是空閒的worker public List<Runnable> shutdownNow() { List<Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(STOP); interruptWorkers(); tasks = drainQueue(); } finally { mainLock.unlock(); } tryTerminate(); return tasks; } //是否已關閉 public boolean isShutdown() { return !isRunning(ctl.get()); } public boolean isTerminating() { int c = ctl.get(); return !isRunning(c) && runStateLessThan(c, TERMINATED); } public boolean isTerminated() { return runStateAtLeast(ctl.get(), 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(); } } protected void finalize() { SecurityManager sm = System.getSecurityManager(); if (sm == null || acc == null) { shutdown(); } else { PrivilegedAction<Void> pa = () -> { shutdown(); return null; }; AccessController.doPrivileged(pa, acc); } } public void setThreadFactory(ThreadFactory threadFactory) { if (threadFactory == null) throw new NullPointerException(); this.threadFactory = threadFactory; } public ThreadFactory getThreadFactory() { return threadFactory; } public void setRejectedExecutionHandler(RejectedExecutionHandler handler) { if (handler == null) throw new NullPointerException(); this.handler = handler; } public RejectedExecutionHandler getRejectedExecutionHandler() { return handler; } public void setCorePoolSize(int corePoolSize) { if (corePoolSize < 0) throw new IllegalArgumentException(); int delta = corePoolSize - this.corePoolSize; this.corePoolSize = corePoolSize; if (workerCountOf(ctl.get()) > corePoolSize) interruptIdleWorkers(); else if (delta > 0) { int k = Math.min(delta, workQueue.size()); while (k-- > 0 && addWorker(null, true)) { if (workQueue.isEmpty()) break; } } } public int getCorePoolSize() { return corePoolSize; } public boolean prestartCoreThread() { return workerCountOf(ctl.get()) < corePoolSize && addWorker(null, true); } public int prestartAllCoreThreads() { int n = 0; while (addWorker(null, true)) ++n; return n; } public boolean allowsCoreThreadTimeOut() { return allowCoreThreadTimeOut; } public void allowCoreThreadTimeOut(boolean value) { if (value && keepAliveTime <= 0) throw new IllegalArgumentException("Core threads must have nonzero keep alive times"); if (value != allowCoreThreadTimeOut) { allowCoreThreadTimeOut = value; if (value) interruptIdleWorkers(); } } public void setMaximumPoolSize(int maximumPoolSize) { if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize) throw new IllegalArgumentException(); this.maximumPoolSize = maximumPoolSize; if (workerCountOf(ctl.get()) > maximumPoolSize) interruptIdleWorkers(); } public int getMaximumPoolSize() { return maximumPoolSize; } public void setKeepAliveTime(long time, TimeUnit unit) { if (time < 0) throw new IllegalArgumentException(); if (time == 0 && allowsCoreThreadTimeOut()) throw new IllegalArgumentException("Core threads must have nonzero keep alive times"); long keepAliveTime = unit.toNanos(time); long delta = keepAliveTime - this.keepAliveTime; this.keepAliveTime = keepAliveTime; if (delta < 0) interruptIdleWorkers(); } public long getKeepAliveTime(TimeUnit unit) { return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS); } public BlockingQueue<Runnable> getQueue() { return workQueue; } public boolean remove(Runnable task) { boolean removed = workQueue.remove(task); tryTerminate(); return removed; } public void purge() { final BlockingQueue<Runnable> q = workQueue; try { Iterator<Runnable> it = q.iterator(); while (it.hasNext()) { Runnable r = it.next(); if (r instanceof Future<?> && ((Future<?>) r).isCancelled()) it.remove(); } } catch (ConcurrentModificationException fallThrough) { for (Object r : q.toArray()) if (r instanceof Future<?> && ((Future<?>) r).isCancelled()) q.remove(r); } tryTerminate(); } public int getPoolSize() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { return runStateAtLeast(ctl.get(), TIDYING) ? 0 : workers.size(); } finally { mainLock.unlock(); } } public int getActiveCount() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { int n = 0; for (Worker w : workers) if (w.isLocked()) ++n; return n; } finally { mainLock.unlock(); } } public int getLargestPoolSize() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { return largestPoolSize; } finally { mainLock.unlock(); } } public long getTaskCount() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { long n = completedTaskCount; for (Worker w : workers) { n += w.completedTasks; if (w.isLocked()) ++n; } return n + workQueue.size(); } finally { mainLock.unlock(); } } public long getCompletedTaskCount() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { long n = completedTaskCount; for (Worker w : workers) n += w.completedTasks; return n; } finally { mainLock.unlock(); } } public String toString() { long ncompleted; int nworkers, nactive; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { ncompleted = completedTaskCount; nactive = 0; nworkers = workers.size(); for (Worker w : workers) { ncompleted += w.completedTasks; if (w.isLocked()) ++nactive; } } finally { mainLock.unlock(); } int c = ctl.get(); String rs = (runStateLessThan(c, SHUTDOWN) ? "Running" : (runStateAtLeast(c, TERMINATED) ? "Terminated" : "Shutting down")); return super.toString() + "[" + rs + ", pool size = " + nworkers + ", active threads = " + nactive + ", queued tasks = " + workQueue.size() + ", completed tasks = " + ncompleted + "]"; } void ensurePrestart() { int wc = workerCountOf(ctl.get()); if (wc < corePoolSize) addWorker(null, true); else if (wc == 0) addWorker(null, false); } private static int runStateOf(int c) { return c & ~CAPACITY; } private static int workerCountOf(int c) { return c & CAPACITY; } private static int ctlOf(int rs, int wc) { return rs | wc; } private static boolean runStateLessThan(int c, int s) { return c < s; } private static boolean runStateAtLeast(int c, int s) { return c >= s; } private static boolean isRunning(int c) { return c < SHUTDOWN; } 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 void advanceRunState(int targetState) { for (; ; ) { int c = ctl.get(); if (runStateAtLeast(c, targetState) || ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c)))) break; } } //嘗試終止執行緒池：如果執行緒池的狀態為SHUTDOWN並且執行緒池和阻塞佇列都為空或者狀態為STOP並且執行緒池為空，則將執行緒池控制狀態轉化為TERMINATED；否則，將中斷一個空閒的worker final void tryTerminate() { for (;;) { // 無限迴圈，確保操作成功 // 獲取執行緒池控制狀態 int c = ctl.get(); if (isRunning(c) || // 執行緒池的執行狀態為RUNNING runStateAtLeast(c, TIDYING) || // 執行緒池的執行狀態最小要大於TIDYING (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) // 執行緒池的執行狀態為SHUTDOWN並且workQueue佇列不為null // 不能終止，直接返回 return; if (workerCountOf(c) != 0) { // 執行緒池正在執行的worker數量不為0 // 僅僅中斷一個空閒的worker interruptIdleWorkers(ONLY_ONE); return; } // 獲取執行緒池的鎖 final ReentrantLock mainLock = this.mainLock; // 獲取鎖 mainLock.lock(); try { if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { // 比較並設定執行緒池控制狀態為TIDYING try { // 終止，鉤子函式 terminated(); } finally { // 設定執行緒池控制狀態為TERMINATED ctl.set(ctlOf(TERMINATED, 0)); // 釋放在termination條件上等待的所有執行緒 termination.signalAll(); } return; } } finally { // 釋放鎖 mainLock.unlock(); } } } private void checkShutdownAccess() { SecurityManager security = System.getSecurityManager(); if (security != null) { security.checkPermission(shutdownPerm); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) security.checkAccess(w.thread); } finally { mainLock.unlock(); } } } private void interruptWorkers() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) w.interruptIfStarted(); } finally { mainLock.unlock(); } } //中斷正在等待任務的空閒worker private void interruptIdleWorkers(boolean onlyOne) { // 執行緒池的鎖 final ReentrantLock mainLock = this.mainLock; // 獲取鎖 mainLock.lock(); try { for (Worker w : workers) { // 遍歷workers佇列 // worker對應的執行緒 Thread t = w.thread; if (!t.isInterrupted() && w.tryLock()) { // 執行緒未被中斷並且成功獲得鎖 try { // 中斷執行緒 t.interrupt(); } catch (SecurityException ignore) { } finally { // 釋放鎖 w.unlock(); } } if (onlyOne) // 若只中斷一個，則跳出迴圈 break; } } finally { // 釋放鎖 mainLock.unlock(); } } private void interruptIdleWorkers() { interruptIdleWorkers(false); } private static final boolean ONLY_ONE = true; final void reject(Runnable command) { handler.rejectedExecution(command, this); } void onShutdown() { } final boolean isRunningOrShutdown(boolean shutdownOK) { int rs = runStateOf(ctl.get()); return rs == RUNNING || (rs == SHUTDOWN && shutdownOK); } private List<Runnable> drainQueue() { 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; } //可能會完成如下幾件任務: //1.原子性的增加workerCount。 //2.將使用者給定的任務封裝成為一個worker，並將此worker新增進workers集合中。 //3.啟動worker對應的執行緒，並啟動該執行緒，執行worker的run方法。 //4.回滾worker的建立動作，即將worker從workers集合中刪除，並原子性的減少workerCount。 private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { // 外層無限迴圈 // 獲取執行緒池控制狀態 int c = ctl.get(); // 獲取狀態 int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && // 狀態大於等於SHUTDOWN，初始的ctl為RUNNING，小於SHUTDOWN ! (rs == SHUTDOWN && // 狀態為SHUTDOWN firstTask == null && // 第一個任務為null ! workQueue.isEmpty())) // worker佇列不為空 // 返回 return false; for (;;) { // worker數量 int wc = workerCountOf(c); if (wc >= CAPACITY || // worker數量大於等於最大容量 wc >= (core ? corePoolSize : maximumPoolSize)) // worker數量大於等於核心執行緒池大小或者最大執行緒池大小 return false; if (compareAndIncrementWorkerCount(c)) // 比較並增加worker的數量 // 跳出外層迴圈 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); // 獲取worker對應的執行緒 final Thread t = w.thread; if (t != null) { // 執行緒不為null // 執行緒池鎖 final ReentrantLock mainLock = this.mainLock; // 獲取鎖 mainLock.lock(); try { // 執行緒池的執行狀態 int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || // 小於SHUTDOWN (rs == SHUTDOWN && firstTask == null)) { // 等於SHUTDOWN並且firstTask為null if (t.isAlive()) // 執行緒剛新增進來，還未啟動就存活 // 丟擲執行緒狀態異常 throw new IllegalThreadStateException(); // 將worker新增到worker集合 workers.add(w); // 獲取worker集合的大小 int s = workers.size(); if (s > largestPoolSize) // 佇列大小大於largestPoolSize // 重新設定largestPoolSize largestPoolSize = s; // 設定worker已被新增標識 workerAdded = true; } } finally { // 釋放鎖 mainLock.unlock(); } if (workerAdded) { // worker被新增 // 開始執行worker的run方法 t.start(); // 設定worker已開始標識 workerStarted = true; } } } finally { if (! workerStarted) // worker沒有開始 // 新增worker失敗 addWorkerFailed(w); } return workerStarted; } private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (w != null) workers.remove(w); decrementWorkerCount(); tryTerminate(); } finally { mainLock.unlock(); } } //在worker退出時呼叫到的鉤子函式。 //worker退出的主要因素: //1.阻塞佇列已經為空，即沒有任務可以運行了。 //2.呼叫了shutDown或shutDownNow函式 //根據是否中斷了空閒執行緒來確定是否減少workerCount的值，並且將worker從workers集合中移除並且會嘗試終止執行緒池 private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) // 如果被中斷，則需要減少workCount decrementWorkerCount(); // 獲取可重入鎖 final ReentrantLock mainLock = this.mainLock; // 獲取鎖 mainLock.lock(); try { // 將worker完成的任務新增到總的完成任務中 completedTaskCount += w.completedTasks; // 從workers集合中移除該worker workers.remove(w); } finally { // 釋放鎖 mainLock.unlock(); } // 嘗試終止 tryTerminate(); // 獲取執行緒池控制狀態 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) // workerCount大於等於min // 直接返回 return; } // 新增worker addWorker(null, false); } } //用於從workerQueue阻塞佇列中獲取Runnable物件，由於是阻塞佇列，所以支援有限時間等待（poll）和無限時間等待（take）。 //在該函式中還會響應shutDown和shutDownNow函式的操作，若檢測到執行緒池處於SHUTDOWN或STOP狀態，則會返回null，而不再返回阻塞佇列中的Runnalbe物件。 private Runnable getTask() { boolean timedOut = false; for (;;) { // 無限迴圈，確保操作成功 // 獲取執行緒池控制狀態 int c = ctl.get(); // 執行的狀態 int rs = runStateOf(c); if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { // 大於等於SHUTDOWN（表示呼叫了shutDown）並且（大於等於STOP（呼叫了shutDownNow）或者worker阻塞佇列為空） // 減少worker的數量 decrementWorkerCount(); // 返回null，不執行任務 return null; } // 獲取worker數量 int wc = workerCountOf(c); boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 是否允許coreThread超時或者workerCount大於核心大小 if ((wc > maximumPoolSize || (timed && timedOut)) // worker數量大於maximumPoolSize && (wc > 1 || workQueue.isEmpty())) { // workerCount大於1或者worker阻塞佇列為空（在阻塞佇列不為空時，需要保證至少有一個wc） if (compareAndDecrementWorkerCount(c)) // 比較並減少workerCount // 返回null，不執行任務，該worker會退出 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; } } } //此函式中會實際執行給定任務（即呼叫使用者重寫的run方法），並且當給定任務完成後，會繼續從阻塞佇列中取任務，直到阻塞佇列為空（即任務全部完成） //在執行給定任務時，會呼叫鉤子函式，利用鉤子函式可以完成使用者自定義的一些邏輯. final void runWorker(Worker w) { // 獲取當前執行緒 Thread wt = Thread.currentThread(); // 獲取w的firstTask Runnable task = w.firstTask; // 設定w的firstTask為null w.firstTask = null; // 釋放鎖（設定state為0，允許中斷） w.unlock(); boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { // 任務不為null或者阻塞佇列還存在任務 // 獲取鎖 w.lock(); if ((runStateAtLeast(ctl.get(), STOP) || // 執行緒池的執行狀態至少應該高於STOP (Thread.interrupted() && // 執行緒被中斷 runStateAtLeast(ctl.get(), STOP))) && // 再次檢查，執行緒池的執行狀態至少應該高於STOP !wt.isInterrupted()) // wt執行緒（當前執行緒）沒有被中斷 wt.interrupt(); // 中斷wt執行緒（當前執行緒） 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; // 增加給worker完成的任務數量 w.completedTasks++; // 釋放鎖 w.unlock(); } } completedAbruptly = false; } finally { // 處理完成後，呼叫鉤子函式 processWorkerExit(w, completedAbruptly); } } protected void beforeExecute(Thread t, Runnable r) { } protected void afterExecute(Runnable r, Throwable t) { } protected void terminated() { } }
執行緒池可以解決兩個不同問題：由於減少了每個任務呼叫的開銷，它們通常可以在執行大量非同步任務時提供增強的效能，並且還可以提供繫結和管理資源（包括執行任務集時使用的執行緒）的方法。
在ThreadPoolExecutor的內部，主要由BlockingQueue和AbstractQueuedSynchronizer對其提供支援，BlockingQueue介面有多種資料結構的實現，如LinkedBlockingQueue、ArrayBlockingQueue等
ThreadPoolExecutor繼承自AbstractExecutorService，AbstractExecuetorService提供了ExecutorService執行方法的預設實現。
類的內部類
ThreadPoolExecutor的核心內部類為Worker，其對資源進行了複用，減少建立執行緒的開銷，還有若干個策略類。內部類的類圖如下：
說明：
可以看到Worker繼承了AQS抽象類並且實現了Runnable介面，是ThreadPoolExecutor的核心內部類。
對於AbortPolicy，用於被拒絕任務的處理程式，它將丟擲 RejectedExecutionException、
CallerRunsPolicy，用於被拒絕任務的處理程式，它直接在 execute 方法的呼叫執行緒中執行被拒絕的任務；如果執行程式已關閉，則會丟棄該任務、
DiscardPolicy，用於被拒絕任務的處理程式，預設情況下它將丟棄被拒絕的任務、
DiscardOldestPolicy，用於被拒絕任務的處理程式，它放棄最舊的未處理請求，然後重試 execute；如果執行程式已關閉，則會丟棄該任務。這些都是拒絕任務提交時的所採用的不同策略。
Worker類
1. 類的繼承關係
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {}
說明：Worker繼承了AQS抽象類，其重寫了AQS的一些方法，並且其也可作為一個Runnable物件，從而可以建立執行緒Thread。
2. 類的屬性
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
// 版本號
private static final long serialVersionUID = 6138294804551838833L;
// worker 所對應的執行緒
final Thread thread;
// worker所對應的第一個任務
Runnable firstTask;
// 已完成任務數量
volatile long completedTasks;
}說明：Worker屬性中比較重要的屬性如下，Thread型別的thread屬性，用來封裝worker（因為worker為Runnable物件），表示一個執行緒；Runnable型別的firstTask，其表示該worker所包含的Runnable物件，即使用者自定義的Runnable物件，完成使用者自定義的邏輯的Runnable物件；volatile修飾的long型別的completedTasks，表示已完成的任務數量。
3. 類的建構函式
Worker(Runnable firstTask) {
// 設定狀態為-1
setState(-1);
// 初始化第一個任務
this.firstTask = firstTask;
// 根據當前worker，初始化執行緒
this.thread = getThreadFactory().newThread(this);
}說明：用於構造一個worker物件，並設定AQS的state為-1，同時初始化了對應的域。
4. 核心函式分析
// 重寫了Runnable的run方法
public void run() {
runWorker(this);
}
// 是否被獨佔,0代表未被獨佔，1代表被獨佔
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) {
// 設定獨佔執行緒為null
setExclusiveOwnerThread(null);
// 設定狀態為0
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()) { // AQS狀態大於等於0並且worker對應的執行緒不為null並且該執行緒沒有被中斷
try {
// 中斷執行緒
t.interrupt();
} catch (SecurityException ignore) {
}
}
}說明：Worker的函式主要是重寫了AQS的相應函式和重寫了Runnable的run函式，重寫的函式比較簡單，具體的可以參見AQS的分析。
拒絕策略
//拒絕策略繼承介面RejectedExecutionHandler，實現了rejectedExecution方法
public static class CallerRunsPolicy implements RejectedExecutionHandler {
//構造器
public CallerRunsPolicy() {
}
//執行呼叫者執行緒中的任務r；如果執行程式已關閉，則會丟棄該任務
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {//執行程式未關閉
r.run();//執行任務r
}
}
}
//拒絕策略繼承介面RejectedExecutionHandler，實現了rejectedExecution方法
public static class AbortPolicy implements RejectedExecutionHandler {
//構造器
public AbortPolicy() {
}
//用於被拒絕任務的處理程式，它總是丟擲 RejectedExecutionException
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
//拒絕策略繼承介面RejectedExecutionHandler，實現了rejectedExecution方法
public static class DiscardPolicy implements RejectedExecutionHandler {
//構造器
public DiscardPolicy() {
}
//用於被拒絕任務的處理程式，不執行任何操作：將丟棄被拒絕的任務
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
//拒絕策略繼承介面RejectedExecutionHandler，實現了rejectedExecution方法
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
//構造器
public DiscardOldestPolicy() {
}
//用於被拒絕任務的處理程式，它放棄最舊的未處理請求，然後重試execute；如果執行程式已關閉，則會丟棄該任務。
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {//執行程式未關閉
e.getQueue().poll();//放棄最舊的未處理請求
e.execute(r);//重試r
}
}
}ThreadPoolExecutor.AbortPolicy、ThreadPoolExecutor.CallerRunsPolicy、ThreadPoolExecutor.DiscardOldestPolicy、ThreadPoolExecutor.DiscardPolicy 均實現了RejectedExecutionHandler 介面：
package java.util.concurrent;
public interface RejectedExecutionHandler {
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
rejectedExecution(Runnable r, ThreadPoolExecutor executor)：當execute方法不能接受某個任務時，可以由 ThreadPoolExecutor 呼叫的方法。
因為超出其界限而沒有更多可用的執行緒或佇列槽時，或者關閉 Executor 時就可能發生這種情況。
在沒有其他替代方法的情況下，該方法可能丟擲未經檢查的 RejectedExecutionException，而該異常將傳播到 execute 的呼叫者。
AbortPolicy的rejectedExecution實現：用於被拒絕任務的處理程式，它總是丟擲 RejectedExecutionException；
CallerRunsPolicy的rejectedExecution實現：執行呼叫者執行緒中的任務r；如果執行程式已關閉，則會丟棄該任務；
DiscardOldestPolicy的rejectedExecution實現：用於被拒絕任務的處理程式，它放棄最舊的未處理請求，然後重試execute；如果執行程式已關閉，則會丟棄該任務；
DiscardPolicy 的rejectedExecution實現：用於被拒絕任務的處理程式，不執行任何操作：將丟棄被拒絕的任務。
類 ThreadPoolExecutor
已實現的介面：
已知子類：
一個 ExecutorService，使用池執行緒中的一個實現執行每個提交的任務，這個實現通常使用 Executors 工廠方法配置。
執行緒池可以解決兩個不同問題：
- 由於減少了每個任務呼叫的開銷，它們通常可以在執行大量非同步任務時提供增強的效能，並且還可以提供繫結和管理資源（包括執行任務集時使用的執行緒）的方法。
- 每個ThreadPoolExecutor 還維護著一些基本的統計資料，如完成的任務數。
為了便於跨大量上下文使用，此類提供了很多可調整的引數和擴充套件鉤子 (hook)。
一般情況，都使用較為方便的 Executors 工廠方法 Executors.newCachedThreadPool()（無界執行緒池，可以進行自動執行緒回收）、Executors.newFixedThreadPool(int)（固定大小執行緒池）和 Executors.newSingleThreadExecutor()（單個後臺執行緒），它們均為大多數使用場景預定義了設定。
如果需要手動配置和調整此類時，則使用以下指導：
- 核心和最大池大小
ThreadPoolExecutor 將根據 corePoolSize（參見 getCorePoolSize()）和 maximumPoolSize（參見 getMaximumPoolSize()）設定的邊界自動調整池大小。
當新任務在方法 execute(java.lang.Runnable) 中提交時：
- 如果執行的執行緒少於 corePoolSize，則建立新執行緒來處理請求，即使其他輔助執行緒是空閒的。
- 如果執行的執行緒多於 corePoolSize 而少於 maximumPoolSize，則僅當佇列滿時才建立新執行緒。
- 如果設定的 corePoolSize 和 maximumPoolSize 相同，則建立了固定大小的執行緒池。
- 如果將 maximumPoolSize 設定為基本的無界值（如 Integer.MAX_VALUE），則允許池適應任意數量的併發任務。
在大多數情況下，核心和最大池大小僅基於構造來設定，不過也可以使用 setCorePoolSize(int) 和 setMaximumPoolSize(int) 進行動態更改。
- 按需構造
預設情況下，即使核心執行緒最初只是在新任務到達時才建立和啟動的，也可以使用方法 prestartCoreThread() 或 prestartAllCoreThreads() 對其進行動態重寫。如果構造帶有非空佇列的池，則可能希望預先啟動執行緒。
- 建立新執行緒
使用 ThreadFactory 建立新執行緒。如果沒有另外說明，則在同一個 ThreadGroup 中一律使用 Executors.defaultThreadFactory() 建立執行緒，並且這些執行緒具有相同的 NORM_PRIORITY 優先順序和非守護程序狀態。通過提供不同的 ThreadFactory，可以改變執行緒的名稱、執行緒組、優先順序、守護程序狀態，等等。如果從 newThread 返回 null 時 ThreadFactory 未能建立執行緒，則執行程式將繼續執行，但不能執行任何任務。
- 保持活動時間
如果池中當前有多於 corePoolSize 的執行緒，則這些多出的執行緒在空閒時間超過 keepAliveTime 時將會終止（參見 getKeepAliveTime(java.util.concurrent.TimeUnit)）。這提供了當池處於非活動狀態時減少資源消耗的方法。如果池後來變得更為活動，則可以建立新的執行緒。也可以使用方法 setKeepAliveTime(long, java.util.concurrent.TimeUnit) 動態地更改此引數。使用 Long.MAX_VALUE TimeUnit.NANOSECONDS 的值在關閉前有效地從以前的終止狀態禁用空閒執行緒。預設情況下，保持活動策略只在有多於 corePoolSizeThreads 的執行緒時應用。但是隻要 keepAliveTime 值非 0， allowCoreThreadTimeOut(boolean) 方法也可將此超時策略應用於核心執行緒。
- 排隊
所有 BlockingQueue 都可用於傳輸和保持提交的任務。可以使用此佇列與池大小進行互動：
- 如果執行的執行緒少於 corePoolSize，則 Executor 始終首選新增新的執行緒，而不進行排隊。
- 如果執行的執行緒等於或多於 corePoolSize，則 Executor 始終首選將請求加入佇列，而不新增新的執行緒。
- 如果無法將請求加入佇列，則建立新的執行緒，除非建立此執行緒超出 maximumPoolSize，在這種情況下，任務將被拒絕。
排隊有三種通用策略：
- 直接提交。工作佇列的預設選項是
SynchronousQueue，它將任務直接提交給執行緒而不保持它們。在此，如果不存在可用於立即執行任務的執行緒，則試圖把任務加入佇列將失敗，因此會構造一個新的執行緒。此策略可以避免在處理可能具有內部依賴性的請求集時出現鎖。直接提交通常要求無界 maximumPoolSizes 以避免拒絕新提交的任務。當命令以超過佇列所能處理的平均數連續到達時，此策略允許無界執行緒具有增長的可能性。 - 無界佇列。使用無界佇列（例如，不具有預定義容量的
LinkedBlockingQueue）將導致在所有 corePoolSize 執行緒都忙時新任務在佇列中等待。這樣，建立的執行緒就不會超過 corePoolSize。（因此，maximumPoolSize 的值也就無效了。）當每個任務完全獨立於其他任務，即任務執行互不影響時，適合於使用無界佇列；例如，在 Web 頁伺服器中。這種排隊可用於處理瞬態突發請求，當命令以超過佇列所能處理的平均數連續到達時，此策略允許無界執行緒具有增長的可能性。 - 有界佇列。當使用有限的 maximumPoolSizes 時，有界佇列（如
ArrayBlockingQueue）有助於防止資源耗盡，但是可能較難調整和控制。佇列大小和最大池大小可能需要相互折衷：使用大型佇列和小型池可以最大限度地降低 CPU 使用率、作業系統資源和上下文切換開銷，但是可能導致人工降低吞吐量。如果任務頻繁阻塞（例如，如果它們是 I/O 邊界），則系統可能為超過您許可的更多執行緒安排時間。使用小型佇列通常要求較大的池大小，CPU 使用率較高，但是可能遇到不可接受的排程開銷，這樣也會降低吞吐量。
- 被拒絕的任務
當 Executor 已經關閉，並且 Executor 將有限邊界用於最大執行緒和工作佇列容量，且已經飽和時，在方法 execute(java.lang.Runnable) 中提交的新任務將被拒絕。execute 方法將呼叫其RejectedExecutionHandler 的 RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) 方法。
下面提供了四種預定義的處理程式策略：
- 在預設的
ThreadPoolExecutor.AbortPolicy中，處理程式遭到拒絕將丟擲執行時RejectedExecutionException。 - 在
ThreadPoolExecutor.CallerRunsPolicy中，執行緒呼叫執行該任務的 execute 本身。此策略提供簡單的反饋控制機制，能夠減緩新任務的提交速度。 - 在
ThreadPoolExecutor.DiscardPolicy中，不能執行的任務將被刪除。 - 在
ThreadPoolExecutor.DiscardOldestPolicy中，如果執行程式尚未關閉，則位於工作佇列頭部的任務將被刪除，然後重試執行程式（如果再次失敗，則重複此過程）。
&nb