JDK 源码解析 —— Executors ExecutorService ThreadPoolExecutor 线程池

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

2. 创建一个单个线程的线程池。任务会被保证顺序执行，因为只有一个工作线程。不像 newFixedThreadPool(1)，这个不保证任务顺序执行。corePoolSize 和 maximumPoolSize 都是 1。

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

3. 创建一个可按需自动扩容的线程池，但是会优先重用线程池中空闲可用的线程。这个类型的线程池将会大大提升执行许多短暂的异步任务的程序。如果线程池中线程都在使用，又有新任务到来，则新增一个线程到线程池。如果线程 60 秒内空闲，则将被终止移除线程池。corePoolSize 为 0，可知一旦线程 60s 空闲就会被移出线程池

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

4. 创建一个在一定延迟时间后调度命令的线程池，或者周期性执行的线程池。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

5. 创建完线程池，然后就该执行任务了。看下内部类 DelegatedExecutorService 里的 execute 方法：可以看到任务执行策略（单线程串行、多线程并行）和任务的具体执行分离，是一个典型的命令模式。

static class DelegatedExecutorService extends AbstractExecutorService {
    private final ExecutorService e;
    DelegatedExecutorService(ExecutorService executor) { e = executor; }
    public void execute(Runnable command) { e.execute(command); }
    public void shutdown() { e.shutdown(); }
    public List<Runnable> shutdownNow() { return e.shutdownNow(); }
    public boolean isShutdown() { return e.isShutdown(); }
    public boolean isTerminated() { return e.isTerminated(); }
    public boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException {
        return e.awaitTermination(timeout, unit);
    }
    public Future<?> submit(Runnable task) {
        return e.submit(task);
    }
    public <T> Future<T> submit(Callable<T> task) {
        return e.submit(task);
    }
    public <T> Future<T> submit(Runnable task, T result) {
        return e.submit(task, result);
    }
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
        return e.invokeAll(tasks);
    }
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                         long timeout, TimeUnit unit)
        throws InterruptedException {
        return e.invokeAll(tasks, timeout, unit);
    }
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException {
        return e.invokeAny(tasks);
    }
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                           long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        return e.invokeAny(tasks, timeout, unit);
    }
}

6. 创建线程池的时候，已经实例化了 ThreadPoolExecutor，所以上面 execute() 方法实际是调用 ThreadPoolExecutor 的 execute：给定的任务可能在未来的某个时刻执行。可能是新建一个线程执行，也可能是线程中原有的线程执行。如果任务不能执行，可能是这个 executor 已经被 shutdown 了，也可能是到达了线程池的执行阈值，任务被拒绝执行处理器处理中。

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)) {
        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);
}

• 执行大量异步的任务时，线程池减少线程的创建来减少开销，提升性能
• 提供了对资源的管理，包括当执行一系列任务时，线程的消耗。每个 ThreadPoolExecutor 也存储有些基本数据，诸如完成的任务数量

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

• 如果正在 running 的线程 (即 worker 线程) 小于 corePoolSize ( workerCountOf(c) < corePoolSize )，尝试创建一个新线程并把传入的 command(任务) 作为它的第一个任务。调用 addWorker(command, true) 自动检查 runState 和 workCount，以便在不能增加线程的时候返回 false。添加成功直接 return，不能添加就要看下面的步骤
• 如果被成功放进队列（if (isRunning(c) && workQueue.offer(command))  workQueue 是一个任务排队的阻塞队列），然后还需要二次检查线程池是否 shut down（上次检查后到这次检查前死亡）。所以我们重新检查状态，如果线程池停止的话，回滚进队操作，或者如果没有工作的线程开启一个新线程（addWorker(null, false);）
• 如果不能让任务进入阻塞队列，然后尝试新增一个线程。如果新增线程失败，可能是线程池 shut down 或者线程池饱和（达到 maxPoolSize），所以接下来抛弃这个任务。

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 &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    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);

                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) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}