并发系列（6）之 ThreadPoolExecutor 详解

时间 2019-11-18

标签并发系列 threadpoolexecutor 详解繁體版

原文原文链接

本文将主要介绍咱们平时最经常使用的线程池 ThreadPoolExecutor ，有可能你平时没有直接使用这个类，而是使用 Executors 的工厂方法建立线程池，虽然这样很简单，可是极可能由于这个线程池发生 OOM ，具体状况文中会详细介绍；html

2、ThreadPoolExecutor 概览

ThreadPoolExecutor 的继承关系如图所示：java

其中：异步

Executor：定义了 executor(Runnable command) 异步接口，可是没有强制要求异步；
ExecutorService：提供了生命周期管理的方法，以及有返回值的任务提交；
AbstractExecutorService：提供了 ExecutorService 的默认实现；

1. 主体结构

public class ThreadPoolExecutor extends AbstractExecutorService {
  private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));  // 状态控制变量，核心
  private final BlockingQueue<Runnable> workQueue;                         // 任务等待队列
  private final HashSet<Worker> workers = new HashSet<Worker>();           // 工做线程集合
  private volatile ThreadFactory threadFactory;       // 线程构造工厂
  private volatile RejectedExecutionHandler handler;  // 拒绝策略
  private volatile long keepAliveTime;                // 空闲线程的存活时间（非核心线程）
  private volatile int corePoolSize;                  // 核心线程大小
  private volatile int maximumPoolSize;               // 工做线程最大容量

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

这里已经能够大体看出 ThreadPoolExecutor 的结构了：源码分析

2. Worker 结构

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
  final Thread thread;  // 持有线程，只有在线程工厂运行失败时为空
  Runnable firstTask;   // 初始化任务，不为空的时候，任务直接运行，不在添加到队列
  volatile long completedTasks;  // 完成任务计数

  Worker(Runnable firstTask) {
    setState(-1);   // AQS 初始化状态
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
  }

  public void run() {
    runWorker(this);  // 循环取任务执行
  }
  ...
  // AQS 锁方法
}

这里很容易理解的是 thread 和 firstTask；可是 Worker 还继承了 AQS 作了一个简易的互斥锁，主要是在中断或者 worker 状态改变的时候使用；具体 AQS 的详细说明能够参考，AbstractQueuedSynchronizer 源码分析；优化

3. ctl 控制变量

ctl 控制变量（简记 c）是一个 AtomicInteger 类型的变量，由两部分信息组合而成（两个值互补影响，又能够经过简单的大小比较判断状态）：this

线程池的运行状态 (runState，简记 rs)，由 int 高位的前三位表示；
线程池内有效线程的数量 (workerCount，简记 wc)，由 int 地位的29位表示；

源码以下：线程

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;  // 线程最大容量

                                                         // 状态量
private static final int RUNNING    = -1 << COUNT_BITS;  // 高位 111，第一位是符号位，1表示负数
private static final int SHUTDOWN   =  0 << COUNT_BITS;  // 高位 000
private static final int STOP       =  1 << COUNT_BITS;  // 高位 001
private static final int TIDYING    =  2 << COUNT_BITS;  // 高位 010
private static final int TERMINATED =  3 << COUNT_BITS;  // 高位 011

private static int runStateOf(int c)     { return c & ~CAPACITY; }  // 运行状态，取前3位
private static int workerCountOf(int c)  { return c & CAPACITY; }   // 线程数量，取后29位
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; } // RUNNING 是负数，必然小于 SHUTDOWN

代码中能够看到状态判断的时候都是直接比较的，这是由于 TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING ；他们的状态变迁关系以下：code

其中：htm

RUNNING：运行状态，可接收新任务；
SHUTDOWN：不可接收新任务，继续处理已提交的任务；
STOP：不接收、不处理任务，中断正在进行的任务
TIDYING：全部任务清空，线程中止；
TERMINATED：钩子方法，执行后的最终状态；

3、ThreadPoolExecutor 源码分析

1. 增长工做线程

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // 这里正常状况下，只要大于SHUTDOWN，则必然不能添加线程；可是这里作了一个优化，
            // 若是线程池还在继续处理任务，则能够添加线程加速处理，
            // SHUTDOWN 表示不接收新任务，可是还在继续处理，
            // firstTask 不为空时，是在添加线程的时候，firstTask 不入队，直接处理
            // workQueue 不为空时，则还有任务须要处理
            // 因此连起来就是 rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()
            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;
            }
        }
        // 此时线程计数，增长成功
      
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {  // 线程建立失败时，直接退出
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    int rs = runStateOf(ctl.get());
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) { // 这里一样检查上面的优化条件
                        if (t.isAlive()) // 若是线程已经启动，则状态错误；
                            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;
    }

2. 提交任务

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))  // 再次检查，状态不是RUNNING的时候，拒绝并移除任务
      reject(command);
    else if (workerCountOf(recheck) == 0)  // 这里是防止状态为SHUTDOWN时，已经添加的任务没法执行
      addWorker(null, false);
  }
  else if (!addWorker(command, false))  // 任务入队失败时，直接添加线程，并运行
    reject(command);
}

流程图以下：blog

因此影响任务提交的因数就有：

核心线程的大小；
是否为阻塞队列；
线程池的大小；

3. 处理任务

工做线程启动以后，首先处理 firstTask 任务（特别注意，这个任务是没有入队的），而后从 workQueue 中取出任务处理，队列为空时，超时等待 keepAliveTime ；

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 ((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);  // 退出时清理
  }
}

private Runnable getTask() {
  boolean timedOut = false; // Did the last poll() time out?

  for (;;) {
    int c = ctl.get();
    int rs = runStateOf(c);

    // 此处保证 SHUTDOWN 状态继续处理任务，STOP 状态中止处理
    if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
      decrementWorkerCount();
      return null;
    }
    int wc = workerCountOf(c);
    boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;  // 是否关闭空闲线程

    if ((wc > maximumPoolSize || (timed && timedOut))  // 若是线程大于最大容量，或者容许关闭，且第一次没取到
      && (wc > 1 || workQueue.isEmpty())) {            // 返回空，最后由 processWorkerExit 清理
      if (compareAndDecrementWorkerCount(c))
        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;
    }
  }
}

4. 中止线程池

public void shutdown() {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    checkShutdownAccess();     // 检查中止权限
    advanceRunState(SHUTDOWN); // 设置线程池状态
    interruptIdleWorkers();    // 设置全部线程中断
    onShutdown();              // hook for ScheduledThreadPoolExecutor
  } finally {
    mainLock.unlock();
  }
  tryTerminate();              // 继续执行等待队列中的任务，完毕后设置 TERMINATED 状态
}

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

能够看到 shutdownNow 只比 shutdown 多了，清空等待队列，可是正在执行的任务仍是会继续执行；

4、拒绝策略

以前提到了，提交任务失败的时候，会执行拒绝操做，在 JDk 中为咱们提供了四种策略：

AbortPolicy：直接抛出 RejectedExecutionException 异常，这是默认的拒绝策略；
CallerRunsPolicy：由调用线程自己运行任务，以减缓提交速度；
DiscardPolicy：不处理，直接丢弃掉；
DiscardOldestPolicy：丢弃最老的任务，并执行当前任务；

5、Executors 工厂方法

另外就是根据线程池参数的不一样，Executors 为咱们提供了4种典型的用法：

SingleThreadExecutor：单线程的线程池，提交任务顺序执行；

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

如代码所示，就是最大线程、核心线程都是1，和无界队列组成的线程池，提交任务的时候就会，直接将任务加入队列顺序执行；

FixedThreadPool：固定线程数量线程池：

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

同 SingleThreadExecutor 同样，只是线程数量由用户决定；

CachedThreadPool：动态调节线程池；

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

这里核心线程为0，队列是 SynchronousQueue 容量为1的阻塞队列，而线程数最大，存活60s，因此有任务的时候直接建立新的线程，超时空闲60s；

ScheduledThreadPool：定时任务线程池，功能同 Timer 相似，具体细节后续还会讲到；

总结

决定线程池运行逻辑的主要有三个变量，核心线程大小，队列容量，线程池容量
最后发现其实 Executors 提供的几种实现，都很典型；可是却容易发生 OOM ，因此最好仍是本身手动建立比较好；