public interface Executor {void execute(Runnable command);
}

public interface RunnableFuture<V> extends Runnable, Future<V> {/*** Sets this Future to the result of its computation* unless it has been cancelled.*/void run();
}

// VarHandle mechanics
private static final VarHandle STATE;
private static final VarHandle RUNNER;
private static final VarHandle WAITERS;
static {try {MethodHandles.Lookup l = MethodHandles.lookup();STATE = l.findVarHandle(FutureTask.class, "state", int.class);RUNNER = l.findVarHandle(FutureTask.class, "runner", Thread.class);WAITERS = l.findVarHandle(FutureTask.class, "waiters", WaitNode.class);} catch (ReflectiveOperationException e) {throw new ExceptionInInitializerError(e);}// Reduce the risk of rare disastrous classloading in first call to// LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773Class<?> ensureLoaded = LockSupport.class;
}protected void set(V v) {if (STATE.compareAndSet(this, NEW, COMPLETING)) {outcome = v;STATE.setRelease(this, NORMAL); // final state
        finishCompletion();}
}

public V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException {if (unit == null)throw new NullPointerException();int s = state;if (s <= COMPLETING &&(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)throw new TimeoutException();return report(s);
}private int awaitDone(boolean timed, long nanos)throws InterruptedException {long startTime = 0L;    WaitNode q = null;boolean queued = false;for (;;) {int s = state;if (s > COMPLETING) {// 已经在终结状态, 返回状态if (q != null)q.thread = null;return s;}else if (s == COMPLETING)// 已经完成了, 但是状态还是 COMPLETING
            Thread.yield();else if (Thread.interrupted()) {// 检查中断
            removeWaiter(q);throw new InterruptedException();}else if (q == null) {// 没有创建 WaitNode 节点, 如果 timed 并且 nanos 大于 0, 创建一个 WaitNodeif (timed && nanos <= 0L)return s;q = new WaitNode();}else if (!queued)// 将新的 WaitNode 放到链表头部, 并尝试 cas 到 waitersqueued = WAITERS.weakCompareAndSet(this, q.next = waiters, q);else if (timed) {final long parkNanos;if (startTime == 0L) { // first timestartTime = System.nanoTime();if (startTime == 0L)startTime = 1L;parkNanos = nanos;} else {long elapsed = System.nanoTime() - startTime;if (elapsed >= nanos) {// 超时了
                    removeWaiter(q);return state;}// park 的时间parkNanos = nanos - elapsed;}// nanos 比较慢, 再次检查, 然后阻塞if (state < COMPLETING)LockSupport.parkNanos(this, parkNanos);}else// 不需要超时的阻塞LockSupport.park(this);}
}

public void run() {if (state != NEW ||!RUNNER.compareAndSet(this, null, Thread.currentThread()))// 不在 NEW 状态, 或者 runner 不为 nullreturn;try {// callable 是在构造器中指定的或用 Executors.callable(runnable, result) 创建的Callable<V> c = callable;if (c != null && state == NEW) {V result;boolean ran;try {result = c.call();ran = true;} catch (Throwable ex) {result = null;ran = false;// 设置异常状态和异常结果
                setException(ex);}if (ran)// 正常完成, 设置完成状态和结果
                set(result);}} finally {// runner must be non-null until state is settled to// prevent concurrent calls to run()runner = null;// state must be re-read after nulling runner to prevent// leaked interruptsint s = state;if (s >= INTERRUPTING)handlePossibleCancellationInterrupt(s);}
}protected void set(V v) {if (STATE.compareAndSet(this, NEW, COMPLETING)) {outcome = v;STATE.setRelease(this, NORMAL); // final state
        finishCompletion();}
}private void finishCompletion() {// assert state > COMPLETING;for (WaitNode q; (q = waiters) != null;) {if (WAITERS.weakCompareAndSet(this, q, null)) {// cas 移除 waiters, 对链表中的每个 Node 的线程 unparkfor (;;) {Thread t = q.thread;if (t != null) {q.thread = null;LockSupport.unpark(t);}WaitNode next = q.next;if (next == null)break;q.next = null; // unlink to help gcq = next;}break;}}// 默认实现什么都没做
    done();callable = null;        // to reduce footprint
}

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {return new FutureTask<T>(callable);
}public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask;
}public <T> T invokeAny(Collection<? extends Callable<T>> tasks)throws InterruptedException, ExecutionException {try {return doInvokeAny(tasks, false, 0);} catch (TimeoutException cannotHappen) {assert false;return null;}
}private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,boolean timed, long nanos)throws InterruptedException, ExecutionException, TimeoutException {if (tasks == null)throw new NullPointerException();int ntasks = tasks.size();if (ntasks == 0)throw new IllegalArgumentException();ArrayList<Future<T>> futures = new ArrayList<>(ntasks);ExecutorCompletionService<T> ecs =new ExecutorCompletionService<T>(this);try {ExecutionException ee = null;final long deadline = timed ? System.nanoTime() + nanos : 0L;Iterator<? extends Callable<T>> it = tasks.iterator();// 提交一个任务到 ecs
        futures.add(ecs.submit(it.next()));--ntasks;int active = 1;for (;;) {// 尝试获取第一个完成的任务的 FutureFuture<T> f = ecs.poll();if (f == null) {// 没有完成的任务if (ntasks > 0) {// 还有没提交的任务, 再提交一个到 ecs--ntasks;futures.add(ecs.submit(it.next()));++active;}else if (active == 0)// 没有还没提交的任务和正在执行的任务了break;else if (timed) {f = ecs.poll(nanos, NANOSECONDS);if (f == null)throw new TimeoutException();nanos = deadline - System.nanoTime();}elsef = ecs.take();}if (f != null) {// 存在已经完成的任务--active;try {// 获取结果并返回return f.get();} catch (ExecutionException eex) {ee = eex;} catch (RuntimeException rex) {ee = new ExecutionException(rex);}}}// 出错, 抛出if (ee == null)ee = new ExecutionException();throw ee;} finally {// 取消所有已经提交的任务
        cancelAll(futures);}
}public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)throws InterruptedException {if (tasks == null)throw new NullPointerException();ArrayList<Future<T>> futures = new ArrayList<>(tasks.size());try {for (Callable<T> t : tasks) {// 提交任务RunnableFuture<T> f = newTaskFor(t);futures.add(f);execute(f);}for (int i = 0, size = futures.size(); i < size; i++) {Future<T> f = futures.get(i);if (!f.isDone()) {// 任务没有完成, get() 等待任务完成try { f.get(); }catch (CancellationException | ExecutionException ignore) {}}}return futures;} catch (Throwable t) {cancelAll(futures);throw t;}
}

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,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;
}

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int COUNT_MASK = (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;// Packing and unpacking ctl
private static int runStateOf(int c)     { return c & ~COUNT_MASK; }
private static int workerCountOf(int c)  { return c & COUNT_MASK; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

/*** Set containing all worker threads in pool. Accessed only when* holding mainLock.*/
private final HashSet<Worker> workers = new HashSet<>();private final class Workerextends AbstractQueuedSynchronizerimplements Runnable{/*** This class will never be serialized, but we provide a* serialVersionUID to suppress a javac warning.*/private static final long serialVersionUID = 6138294804551838833L;final Thread thread;Runnable firstTask;volatile long completedTasks;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);}...
}

final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {while (task != null || (task = getTask()) != null) {// task 不为 null 或 获取到了需要执行的任务; getTask() 会阻塞, 并在线程需要退出时返回 null
            w.lock();// 检查线程池状态和线程的中断状态, 如果被中断, 代表线程池正在 STOPif ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())// 重新设置中断状态
                wt.interrupt();try {// 执行前的钩子
                beforeExecute(wt, task);try {// 执行任务
                    task.run();// 执行后的钩子afterExecute(task, null);} catch (Throwable ex) {// 执行后的钩子
                    afterExecute(task, ex);throw ex;}} finally {// 更新状态, 准备处理下一个任务task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {// 处理 Worker 的退出
        processWorkerExit(w, completedAbruptly);}
}

private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();if (runStateAtLeast(c, SHUTDOWN)&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {// 线程池已经处于 SHUTDOWN 状态, 并且不在需要线程 (线程池已经处于 STOP 状态 或 workQueue 为空)
            decrementWorkerCount();return null;}int wc = workerCountOf(c);// 是否需要剔除超时的 workerboolean timed = allowCoreThreadTimeOut || wc > corePoolSize;if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {// 需要剔除当前 worker, 尝试调整 workerCountif (compareAndDecrementWorkerCount(c))// 成功 返回 nullreturn null;continue;}try {// 阻塞获取任务Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;// 设置超时标记, 下一次循环中检查是否需要返回 nulltimedOut = true;} catch (InterruptedException retry) {// 被中断, 设置超时标记, 下一次循环中检查是否需要返回 nulltimedOut = false;}}
}

private void processWorkerExit(Worker w, boolean completedAbruptly) {if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// 更新线程池完成的任务数量completedTaskCount += w.completedTasks;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)// 不需要在添加新 workerreturn;}// 尝试添加新的 workeraddWorker(null, false);}
}

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();if (workerCountOf(c) < corePoolSize) {// 有效线程数量小于 corePoolSize 尝试调用 addWorker 来增加一个线程(在 addWorker 方法中使用 corePoolSize 来检查是否需要增加线程), 使用 corePoolSize 作为, 并把 command 作为新线程的第一个任务if (addWorker(command, true))return;// 调用失败, 重新获取状态c = ctl.get();}if (isRunning(c) && workQueue.offer(command)) {// 线程池仍然在运行, 将 command 加入 workQueue 成功, 再次检查状态, 因为此时线程池状态可能已经改变, 按照新的状态拒绝 command 或尝试添加新的线程int recheck = ctl.get();if (! isRunning(recheck) && remove(command))// 不再是运行中状态, 尝试从队列移除 command(还会尝试将线程池状态转换为 TERMINATED), 拒绝command
            reject(command);else if (workerCountOf(recheck) == 0)// 有效线程数量为 0 , 创建新的线程, 在 addWorker 方法中使用 maximumPoolSize 来检查是否需要增加线程addWorker(null, false);}else if (!addWorker(command, false))// 将任务放入队列失败或线程池不在运行状态, 并且尝试添加线程失败(此时线程池已经 shutdown 或饱和), 拒绝任务
        reject(command);
}

private boolean addWorker(Runnable firstTask, boolean core) {// 重试标签
    retry:for (int c = ctl.get();;) {// 获取最新的状态, 检查状态if (runStateAtLeast(c, SHUTDOWN)&& (runStateAtLeast(c, STOP)|| firstTask != null|| workQueue.isEmpty()))// 如果线程池状态已经进入 SHUDOWN, 并且不再需要工作线程(已经进入 STOP 状态 或 firstTask 不为 null 或 workQueue为空) 返回 falsereturn false;for (;;) {if (workerCountOf(c)>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))// 有效线程数量大于边界条件, 返回 falsereturn false;if (compareAndIncrementWorkerCount(c))// 调整 workerCount, break retry, 退出外部循环break retry;c = ctl.get();  // Re-read ctlif (runStateAtLeast(c, SHUTDOWN))// 因为状态变化导致 CAS 失败, continue retry, 重试外部循环continue retry;// 由于 workerCount 改变导致 CAS 失败, 重试内嵌循环
        }}boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {// 新建 Workerw = new Worker(firstTask);final Thread t = w.thread;if (t != null) {// threadFactory 成功创建了线程final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int c = ctl.get();// 重新检查状态if (isRunning(c) ||(runStateLessThan(c, STOP) && firstTask == null)) {// 线程池在 RUNNING 状态 或 需要线程(线程池还不在 STOP 状态 并且 firstTask 为 null)// 检查线程是否可启动if (t.isAlive()) throw new IllegalThreadStateException();// 将 worker 添加到 workers
                    workers.add(w);// 更新 largestPoolSizeint s = workers.size();if (s > largestPoolSize)largestPoolSize = s;// 更新 worker 添加的标记workerAdded = true;}} finally {mainLock.unlock();}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();try {// 从 workers 中移除 workerif (w != null)workers.remove(w);// 调整 workerCount()
        decrementWorkerCount();// 尝试将线程池状态改变为 TERMINATED
        tryTerminate();} finally {mainLock.unlock();}
}

public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();// 如果线程池状态还没有达到SHUTDOWN, 将线程池状态改为 SHUTDOWN
        advanceRunState(SHUTDOWN);// 中断空闲的工作者线程
        interruptIdleWorkers();// 钩子onShutdown(); // hook for ScheduledThreadPoolExecutor} finally {mainLock.unlock();}// 尝试转换状态到终止
    tryTerminate();
}public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();// 如果线程池状态还没有达到 STOP, 将线程池状态改为 STOP
        advanceRunState(STOP);// 中断所有 worker
        interruptWorkers();// 获取任务队列中的任务, 并将这些任务从任务队列中删除tasks = drainQueue();} finally {mainLock.unlock();}// 尝试转换状态到终止
    tryTerminate();return tasks;
}public boolean awaitTermination(long timeout, TimeUnit unit)throws InterruptedException {long nanos = unit.toNanos(timeout);final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// 等待线程池终止或超时while (runStateLessThan(ctl.get(), TERMINATED)) {if (nanos <= 0L)// 剩余时间小于 0 , 超时return false;nanos = termination.awaitNanos(nanos);}return true;} finally {mainLock.unlock();}
}

final void tryTerminate() {for (;;) {int c = ctl.get();if (isRunning(c) ||runStateAtLeast(c, TIDYING) ||(runStateLessThan(c, STOP) && ! workQueue.isEmpty()))// 状态不需要改变 (处于 RUNNING 状态 或 已经处于 TIDYING 状态 或 (还没到达 STOP 状态, 并且 workQueue 不为空))return;if (workerCountOf(c) != 0) { // Eligible to terminate// 中断一个空闲的 worker, 以传播关闭状态到工作线程
            interruptIdleWorkers(ONLY_ONE);return;}final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {// 将状态成功更新为 TIDYINGtry {// 默认实现没有做任何事情
                    terminated();} finally {// 将线程池状态更新为 TERMINATEDctl.set(ctlOf(TERMINATED, 0));// 唤醒等待终结的线程
                    termination.signalAll();}return;}} finally {mainLock.unlock();}// else retry on failed CAS
    }
}