通过使用标准化的，广泛测试过的并发构建模块，可以消除很多潜在的线程编程风险，诸如死锁、饥饿、竞争条件(race conditions)和过度的上下文切换。

看了一下 ThreadPoolExecutor 源码的实现，觉得真的是赏心悦目。写的很漂亮，至少在我眼里是很漂亮的。

ThreadPoolExecutor 是并发实用工具(utilities)中 Task scheduling framework 里面的默认的线程池实现，具有灵活性和可扩展性。并发实用工具还有 Fork/join framework、Concurrent collections、Atomic varibles、Synchronizers、Locks 和 Nanosecond-granularity timing。有时间再去翻翻其他相关的实现，啊哈哈哈。
我阅读源码的 jdk 版本为 1.8.0_131，强烈建议先读 api文档，再阅读 jdk 的代码，会让你的思路变得清晰。

类图

简单基本的关系图，继承关系非常清晰。其中 Future 属于任务提交后的异步结果，Executors 是创建 ExecutorService、Callable 等的工厂，RejectedExecutionException 是任务不能被接收时抛出的异常。
接下来简单看一下接口和抽象类。

Executor

1
2
3

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

Executor 接口将任务提交和线程使用的细节、调度解耦，也是该并发框架的核心，只有一个 void execute(Runnable command) 方法，通常负责创建线程。

ExecutorService

public interface ExecutorService extends Executor {
    // 终止线程池
    void shutdown();
    
    List<Runnable> shutdownNow();

    boolean isShutdown();

    boolean isTerminated();

    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;
    // 提交任务
    <T> Future<T> submit(Callable<T> task);

    <T> Future<T> submit(Runnable task, T result);

    Future<?> submit(Runnable task);
    // 集中处理任务
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;

    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

ExecutorService 更像是对 Executor 的补充，提供终止线程池、提交任务和集中处理任务。

AbstractExecutorService

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

AbstractExecutorService 除了实现 submit() 和 invoke() 相关的方法外，还新增了两个装饰任务的方法 newTaskFor()。这两个方法在新增任务的时候会被调用。

简单使用

官方推荐使用 Executors 创建线程池，默认创建的是 ThreadPoolExecutor，当然也可以继承 ThreadPoolExecutor，并使用自己定义的线程池。为了简便，以下实例直接从 api 文档 中搬运过来。

class NetworkService implements Runnable {
  private final ServerSocket serverSocket;
  private final ExecutorService pool;

  public NetworkService(int port, int poolSize)
      throws IOException {
    serverSocket = new ServerSocket(port);
    pool = Executors.newFixedThreadPool(poolSize);
  }

  public void run() { // run the service
    try {
      for (;;) {
        pool.execute(new Handler(serverSocket.accept()));
      }
    } catch (IOException ex) {
      pool.shutdown();
    }
  }
}

class Handler implements Runnable {
  private final Socket socket;
  Handler(Socket socket) { this.socket = socket; }
  public void run() {
    // read and service request on socket
  }
}

终止线程池的方法

void shutdownAndAwaitTermination(ExecutorService pool) {
  pool.shutdown(); // Disable new tasks from being submitted
  try {
    // Wait a while for existing tasks to terminate
    if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
      pool.shutdownNow(); // Cancel currently executing tasks
      // Wait a while for tasks to respond to being cancelled
      if (!pool.awaitTermination(60, TimeUnit.SECONDS))
          System.err.println("Pool did not terminate");
    }
  } catch (InterruptedException ie) {
    // (Re-)Cancel if current thread also interrupted
    pool.shutdownNow();
    // Preserve interrupt status
    Thread.currentThread().interrupt();
  }
}

使用起来很简单，只要将需要执行的任务提交给线程池即可，线程管理、调度等都会帮你搞定。但是，只有这样的功能足够优秀么？

首先有个概念，线程池“雇佣了”多名工人去执行任务。工人指的就是线程池中的线程，任务指的是原本需要创建的线程（现在交给了线程池执行）。所以节省了线程频繁创建和销毁的开销，并且可以控制线程占用资源。
ThreadPoolExecutor 还提供核心工人数的设置、预创建工人、创建新工人的 factory、工人存活时间设置、数据结构的设置、四种异常处理策略、hook、清除任务等功能。具有不错的灵活性和健壮性。

说了这么多，最想了解的还是对线程管理的那一块，很明显，就是 execute() 方法了。其余的设计放到后面再进行阅读。所以，接下来阅读的是 execute() 方法。

execute 方法

execute 方法负责创建线程和安排任务，可以说是一个调度者，考虑什么时候要加工人，安排任务需要注意什么。也是线程管理的关键。

代码

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
    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);
}

execute() 方法做的是添加工人或将任务添加到工作队列中。其中的逻辑如下：

如果工人数量小于核心线程数量，则添加工人，并将任务作为初始任务给这个工人
否则将任务添加到工作队列中（workQueue.offer(command)），并做二次校验（校验是否应该添加工人）
尝试添加核心线程数以外的工人，如果仍失败，则拒绝本次任务的提交

看到这里，是不是觉得几行代码就将核心工程完成了！那么，任务交给谁去完成了呢？这就需要看一下内部类 Worker 了，所以真正干活的是工人。

工人 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;

    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }

    public void lock()        { acquire(1); }
    public void unlock()      { release(1); }
    ...
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            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);
    }
}

变量	二进制	十进制
COUNT_BITS		29
CAPACITY	00011111111111111111111111111111
~CAPACITY	11100000000000000000000000000000
RUNNING	11100000000000000000000000000000	-536870912
SHUTDOWN	00000000000000000000000000000000	0
STOP	00100000000000000000000000000000	536870912
TIDYING	01000000000000000000000000000000	1073741824
TERMINATED	01100000000000000000000000000000	1610612736

阅读源码系列--线程池 ThreadPoolExecutor

类图

Executor

ExecutorService

AbstractExecutorService

简单使用

execute 方法

代码

工人 Worker

防止争抢任务

线程池状态和工人数

打包成一个整数

线程池的状态

我们来聊聊设计

ThreadPoolExecutor 的构造方法

protected 方法

newTaskFor

beforeExecute、afterExecute 和 terminated

finalize

如何去设计