相比1.6，1.7有些变化：
1、 添加了一个TIDYING状态。这个状态是介于STOP和TERMINATED之间的。假设运行完terminated钩子函数后状态就变成TERMINATED了；
2、 内部类Worker继承了AQS类作为一个独享锁，在执行每一个任务前会获取自己的锁。
3、 runState和poolSize两个字段被合并成一个原子字段ctl了，不再使用mainLock保护了。
一、成员变量介绍
public class ThreadPoolExecutor extends AbstractExecutorService { /** * ctl字段事实上表示两个含义：runState和workerCount(近似1.6中的poolSize) * int类型。高3位表示runState，低29位表示workerCount。 眼下这个版本号也就限 * 制了线程个数不会超过2^29-1。 * RUNNING: 能接受新的任务且能处理队列里的请求 * SHUTDOWN: 不能接受新的任务可是能处理队列里的请求 * STOP: 不能接受新的任务、不能处理队列里的请求。workers会被interrupt * TIDYING: 全部的线程都已经terminated了，正准备调用terminated()方法 * TERMININATED: terminated()方法已经调用结束了 * * RUNNING->SHUTDOWN: 调用shutdown方法 * (RUNNING/SHUTDOWN)>STOP: 调用shutdownNow方法 * SHUTDOWN->TIDYING: 当workers和queue都空的时候 * STOP->TIDYING: 当workers为空的时候 * TIDYING->TERMINATED: 当terminated方法调用结束的时候。 * awaitTermination()直到状态为TERMINATED时才会返回。 * */ private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3; private static final int CAPACITY = (1 << COUNT_BITS) - 1; // runState is stored in the high-order bits 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; // 取ctl的高三位。获取runState(执行状态) private static int runStateOf(int c) { return c & ~CAPACITY; } // 取ctl的低29位，获取workerCount(worker的数量) private static int workerCountOf(int c) { return c & CAPACITY; } // 把runState和workerCount合并成ctl，上面两个函数的反操作 private static int ctlOf(int rs, int wc) { return rs | wc; }
二、execute函数
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * 三步走： * 1. 假设RUNNING的线程数目小于corePoolSize，直接调用addWorker方法 * 启动一个新线程。addWorker函数会检查runState和workerCount。假设不 * 须要新建一个thread就会返回false了 * * 2. 假设任务被成功的放入了workQueue，我们仍然须要做个double-check * 由于调用完isRunning(c)后池中的线程可能都退出了或者线程池被shut * down了。又一次检查状态看是要remove掉新来的任务还是创建一个新线程来执 * 行(假设没有活动的线程了) * * 3. 假设放入workQueue失败了，我们尝试创建一个新worker。 假设失败了， * 说明线程池被关闭了或者饱和了(超过最大值了)。就直接拒了。 * */ int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { // addWorker有可能会失败，失败后又一次获取状态并继续往下走 if (addWorker(command, true)) return; c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 假设isRunning(c)&&workQueue.offer中间并发发生了shutdown，须要remove // 掉刚放入workQueue的command任务。注意:此时假设有一个worker刚运行完一个task // 然后从workQueue获取下一个task时，这里的remove就会失败了。 if (! isRunning(recheck) && remove(command)) reject(command); // 假设是RUNNING状态可是没有可工作的线程。须要直接new一个 else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); }
execute函数大体思路和1.6一致，就三种情况：
① 当前线程池中线程数目小于corePoolSize，直接new一个thread。
② 当先线程池数据大于corePoolSize，则放入workQueue中；
③ 假设workQueue满了且线程池中线程数小于maximumPoolSize，则new一个thread。
private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c);// 假设被shutdown了。一般就直接返回false。可是须要排除一个特例情况：当线程池状// 态是shutdown。但workQueue不空且workers空了,会调用addWorker(null,false)// 方法创建一个线程处理workQueue里的任务,这时不能直接返回false。 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { int wc = workerCountOf(c);// 假设当前workers数目大于CAPACITY或者大于用户设置了。直接返回false if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl // 假设不过workerCount变化了，那么继续内层的循环；假设连runState也变化了， // 则要又一次继续外层的循环。
if (runStateOf(c) != rs) continue retry; } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { final ReentrantLock mainLock = this.mainLock; w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int c = ctl.get(); int rs = runStateOf(c); // 再次检查runState的状态，假设是RUNNING或者SHUTDOWN可是firstTask不空，则 // 把new出来的worker放入workers中。 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 创建worker成功后直接启动线程了 t.start(); workerStarted = true; } } } finally { if (! workerStarted) // 创建失败要做清理操作 addWorkerFailed(w); } return workerStarted; }
addWorker函数尝试新建一个thread来执行传递给它的task。当线程池被STOP或SHUTDOWN或threadFactory返回null时或者OOM时。会返回false并做对应的清理。整个过程分为两步：1、尝试设置workerCount，成功了就到步骤2；2、尝试创建一个worker并增加到workers里。
private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (w != null) workers.remove(w); decrementWorkerCount(); tryTerminate(); } finally { mainLock.unlock(); } } addWorkerFailed函数做些清理操作：1、把创建的worker从workers中删除；2、把workerCount减1；3、检查能否够terminated线程池，防止这个worker的存在导致运行awaitTermination操作的client线程堵塞了。 final void tryTerminate() { for (;;) { int c = ctl.get();// 假设是以下三种情况直接返回：// 1.RUNNING状态; 2.runState>=TIDYING。说明有其它线程运行了tryTerminate操// 作; 3.SHUTDOWN状态且workQueue不空 if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return;// 假设workerCount大于0。则中断一个空暇的worker,就返回了。为啥仅仅中断一个呢？// 由于worker线程退出时也会调用tryTerminate方法(一个接一个的传播) if (workerCountOf(c) != 0) { // Eligible to terminate interruptIdleWorkers(ONLY_ONE); return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try {// 走到这里说明workers数量为0了，尝试把线程池状态改成TIDYING并调用terminated// 函数->状态再设置成TERMINATED。 假设设置TIDYING失败，则继续循环。 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { try { terminated(); } finally { // terminated函数抛异常也须要运行以下的操作。 ctl.set(ctlOf(TERMINATED, 0)); termination.signalAll(); } return; } } finally { mainLock.unlock(); } // else retry on failed CAS } }
tryTerminate函数尝试TERMINATED线程池（当a、SHUTDOWN且queue和pool都空；b、STOP且queue为空了）。假设workers不为0，则中断随意一个空暇的worker后直接返回。否则：首先，将线程池状态改成TIDYING；其次，调用用户的钩子函数terminated;最后，将状态设置成TERMINATED。
private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) { Thread t = w.thread;// 假设tryLock成功，就说明这个worker是空暇的。 if (!t.isInterrupted() && w.tryLock()) { try { t.interrupt(); } catch (SecurityException ignore) { } finally { w.unlock(); } } if (onlyOne) // 假设仅仅中断一个就break，仅仅有tryTerminate函数中使用到这样的情况。 break; } } finally { mainLock.unlock(); } }
interruptIdleWorkers函数依据onlyOne參数决定中断一个或全部空暇的workers(这些workers都堵塞在getTask方法中)。
三、shutdown函数
public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try {// 检查调用者是否有权限运行shutdown checkShutdownAccess();// 将线程池的状态改成SHUTDOWN advanceRunState(SHUTDOWN);// 中断全部空暇的workers interruptIdleWorkers(); onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); }// 尝试终止线程池 tryTerminate(); } shutdown函数就运行几步：把状态改成SHUTDOWN。中断全部空暇的workers，调用onShutdown钩子函数，最后调用tryTerminate尝试终止线程池。
private void advanceRunState(int targetState) { for (;;) { int c = ctl.get(); if (runStateAtLeast(c, targetState) || ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c)))) break; } } advanceRunState函数将线程池的状态改成指定状态值。假设如今状态值比target值大就直接返回。targeState的值是SHUTDOWN或者STOP，不能是TIDYING或者TERMINATED（这两种状态须要调用tryTerminate函数设置）。
四、shutdownNow函数
public ListshutdownNow() { List tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try {// 检查调用者是否有权限运行关闭 checkShutdownAccess();// 将线程池的状态改成STOP advanceRunState(STOP);// 和shutdown不同，这里中断全部的worker线程 interruptWorkers();// 删除workQueue里的任务并返回任务列表 tasks = drainQueue(); } finally { mainLock.unlock(); }// 尝试终止线程池 tryTerminate(); return tasks; }
shutdownNow函数会中断全部的worker线程，删除workQueue里的任务，最后尝试终止线程池并返回workQueue里的任务。
private void interruptWorkers() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try {// 中断全部的worker线程 for (Worker w : workers) w.interruptIfStarted(); } finally { mainLock.unlock(); } } private ListdrainQueue() { BlockingQueue q = workQueue; List taskList = new ArrayList (); q.drainTo(taskList); if (!q.isEmpty()) { for (Runnable r : q.toArray(new Runnable[0])) { if (q.remove(r)) taskList.add(r); } } return taskList; }
五、Worker内部类
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** Thread this worker is running in. Null if factory fails. */ final Thread thread; /** Initial task to run. Possibly null. */ Runnable firstTask; /** Per-thread task counter */ volatile long completedTasks; /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */ Worker(Runnable firstTask) {// 初始值为-1，防止worker还没启动就被interrupt了；在start開始时会将状态改成0 setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); }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; }// 參数1没有意义，是独占锁 public void lock() { acquire(1); } public boolean tryLock() { return tryAcquire(1); } public void unlock() { release(1); } public boolean isLocked() { return isHeldExclusively(); } Worker类主要维护着中断的管理和其它操作(runWorker函数)，继承了AQS类实现了一个不可重入的Lock。在获取到一个任务后，准备运行前首先要获取这个锁。
同一时候。在中断空暇的worker时也要先获取到这个锁。
public void run() { runWorker(this); } final void runWorker(Worker w) { Thread wt = Thread.currentThread();// 有时我们不想从workQueue取第一个任务，直接运行刚提交的任务 Runnable task = w.firstTask; w.firstTask = null;// 把state设置成0，同意中断 w.unlock(); // allow interrupts boolean completedAbruptly = true; try {// 进入循环了 while (task != null || (task = getTask()) != null) { w.lock(); // 假设是STOP状态，须要保证线程是被中断了的; // 假设不是须要清空中断状态,可是须要又一次检查下状态防止在清除 // 中断时发生了shutdownNow 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 { processWorkerExit(w, completedAbruptly); } } runWorker函数循环从workQueue里获取task并运行，可是须要注意下面几个问题：1.假设不想从workQueue里获取第一个任务运行，那就给worker.firstTask赋值。2、假设getTask获取的值为null，或者你的task里抛异常了，那循环就退出了，然后worker线程也就退出了。3、在运行任务前先要获取worker的锁，这里防止中断正在运行的线程。4、假设你的钩子函数beforeExecute函数抛异常了，那么你的任务就不会被运行了，worker线程也会退出。5、假设task.run方法抛出Runtime或Error异常，会原样抛出。假设是Throwable，则会包装成一个Error抛出，抛出异常前会运行afterExecute钩子函数。最后线程会退出。6、假设afterExecute钩子函数抛出异常。那么worker线程也会退出。
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // 假设SHUTDOWN且workQueue为空，或者STOP了。worker线程直接退出 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; }// 是否要回收这个worker线程？ boolean timed; // Are workers subject to culling? for (;;) { int wc = workerCountOf(c); timed = allowCoreThreadTimeOut || wc > corePoolSize;// 假设还没有超时过(循环第一次运行到这里)直接break if (wc <= maximumPoolSize && ! (timedOut && timed)) break;// 否则。假设线程数大于最大限制或者已经超时过了说明这个worker线程要准备退出了// 先设置workerCount-1，成功的话直接退出。否则，看下runState是否和rs一样，如// 果一样就在内部循环，不一样就要到外部循环 if (compareAndDecrementWorkerCount(c)) return null; c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } try {// 无限堵塞或超时堵塞 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; // 没有获取到task肯定是超时了 timedOut = true; } catch (InterruptedException retry) { // 假设被中断了，不能算作超时 timedOut = false; } } }
getTask函数是从workQueue里获取一个task，有两种策略（无限堵塞或者超时，详细要看client的配置）。
假设这个函数返回了null，那么worker线程就会退出了。退出的原因不外乎下面几种：
1. 当前线程池中worker数量大于maximumPoolSize了。
2. 线程池被STOP了(workQueue.poll/take时会捕获到InterruptedException异常)；
3. 线程池被SHUTDOWN了且workQueue为空(workQueue.poll/take时会捕获到InterruptedException异常)；
4. 获取task超时了(timedOut)&&(timed)。
private void processWorkerExit(Worker w, boolean completedAbruptly) {// 用户的函数抛异常了，须要调整workerCount的值，由于worker线程准备退出了 if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount();// 做些统计操作(bookkeeping) final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { completedTaskCount += w.completedTasks; workers.remove(w); } finally { mainLock.unlock(); }// 尝试终止线程池 tryTerminate(); int c = ctl.get();// 假设是RUNNING或SHUTDOWN状态，要看下workQueue是否为空,// 不能直接退出。 假设workQueue不空，至少要保留1或corePoolSize个 // 线程(看allowCoreThreadTimeOut配置)。少于这个数目，就须要通过 // addWorker(null,false)方法补充新的线程进来。
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); } }
processWorkerExit函数是在runWork循环退出后做的清理和bookkeeping(应该就是指completedTaskCount等变量的操作吧)操作。
completedAbruptly參数的含义是指用户的函数是否抛异常了(before/after/run等)。注意下函数最后会依据线程池的状态和配置决定是否新建一个worker线程。