zoukankan html css js c++ java

Java并发之ThreadPoolExecutor源码解析

Java并发之ThreadPoolExecutor源码解析（123）

线程池

假设我们编写了一个Servlet应用，当用户通过浏览器发起一个请求到达我们服务器时，传统的Servlet应用一般针对一个用户请求创建一个线程去执行请求，等到请求执行完毕后，再销毁线程。这种设计在用户量几百或者几千的情况下一般不会有什么大问题，但是如果我们的用户量上达几万甚至几十万几百万，频繁的创建、销毁线程，将会给服务器带来巨大的开销，甚至会出现OOM（Out Of Memory）异常。因此，为了节省资源的消耗，提高资源的利用率，引出了线程池化技术。

线程池会维护若干线程，等待任务的到来，避免重复创建、销毁线程造成的消耗，提高任务的响应速度，不需要创建线程就可以立即执行任务，使用线程池可以进行统一的分配、调优和监控，避免无节制的创建线程，提高系统的稳定性。

当然，线程池也并非十全十美，它也有力不从心的场景，例如：线程池适合生命周期较短的任务，不适合耗时较长的任务；线程池无法设置任务的优先级，也无法单独启动或者终止某个线程。

现在，我们对比一下线程池执行任务和创建线程执行任务的优势。在ThreadTest中，我们声明了一个线程安全的list，并创建10000个线程并发往list添加随机值，最后我们等待所有线程执行完毕，打印程序的执行时间和list的大小。

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;
 
public class ThreadTest {
 
    public static void main(String[] args) throws InterruptedException {
        Long start = System.currentTimeMillis();
        final Random random = new Random();
        final List<Integer> list = Collections.synchronizedList(new ArrayList<>());
        final List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            Thread thread = new Thread(() -> list.add(random.nextInt()));
            thread.start();
            threads.add(thread);
        }
        for (Thread thread : threads) {
            thread.join();
        }
        System.out.println("时间：" + (System.currentTimeMillis() - start));
        System.out.println("大小：" + list.size());
    }
}

执行结果：

时间：882
大小：10000

可以看到list的长度为10000，程序执行了882毫秒。

下面，我们用线程池的方式来执行相同的逻辑，我们声明一个线程池executorService，并往线程池中提交10000个任务，每个任务都向list添加一个随机值：

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
 
public class ThreadPoolTest {
 
    public static void main(String[] args) throws InterruptedException {
        Long start = System.currentTimeMillis();
        final Random random = new Random();
        final List<Integer> list = Collections.synchronizedList(new ArrayList<>());
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 10000; i++) {
            executorService.execute(() -> list.add(random.nextInt()));
        }
        executorService.shutdown();
        executorService.awaitTermination(1, TimeUnit.DAYS);
        System.out.println("时间：" + (System.currentTimeMillis() - start));
        System.out.println("大小：" + list.size());
    }
}

执行结果：

时间：52
大小：10000

可以看到线程池的执行时间相比创建线程，大大缩短。

下面，我们就从源码的角度，来剖析线程池的工作原理。ThreadPoolExecutor是java.util.concurrent包下提供的线程池实现类，下图是ThreadPoolExecutor类的继承关系，我们将从上至下逐个分析ThreadPoolExecutor的父类：Executor、ExecutorService、AbstractExecutorService。

Executor

我们先来看下Executor接口定义：

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

Executor允许我们提交若干待执行的任务，我们不再像以前一样用new Thread(new RunnableTask()).start()的方式启动一个线程去执行RunnableTask的run()方法，取而代之的是用Executor的实现类去执行，比如：

Executor executor = anExecutor;
executor.execute(new RunnableTask1());
executor.execute(new RunnableTask2());
...

Executor提供了一种新的方式，我们只需提交任务，Executor自身负责如何调度线程来执行任务。Executor并不要求任务的执行必须是异步的，也可以在提交完任务后，同步执行任务：

class DirectExecutor implements Executor {
    public void execute(Runnable r) {
        r.run();
    }
}

通常情况下，Executor会将提交过来的任务放在另一个线程执行，而不是通过调用线程来执行：

class ThreadPerTaskExecutor implements Executor {
    public void execute(Runnable r) {
        new Thread(r).start();
    }
}

一些Executor接口的实现在调度线程执行任务时会添加一些限制，比如我们可以以代理模式的思想来封装Executor：

class SerialExecutor implements Executor {
  final Queue<Runnable> tasks = new ArrayDeque<Runnable>();
  final Executor executor;
  Runnable active;
 
  SerialExecutor(Executor executor) {
    this.executor = executor;
  }
 
  public synchronized void execute(final Runnable r) {
    tasks.offer(new Runnable() {
      public void run() {
        try {
          r.run();
        } finally {
          scheduleNext();
        }
      }
    });
    if (active == null) {
      scheduleNext();
    }
  }
 
  protected synchronized void scheduleNext() {
    if ((active = tasks.poll()) != null) {
      executor.execute(active);
    }
  }
}

在创建SerialExecutor对象时，会要求传入一个executor对象，执行任务并不是SerialExecutor对象本身，SerialExecutor并不执行任务，只是将任务缓存到队列tasks，而executor才是真正负责执行任务。

ExecutorService

ExecutorService扩展了Executor，ExecutorService不但具备Executor执行任务的能力，我们还可以关闭ExecutorService，这将使ExecutorService拒绝接受新提交的任务。ExecutorService.submit(...)方法是基于Executor.execute(Runnable command)封装的，execute方法没有任何返回，而submit会返回一个Future对象，通过Future对象我们可以取消任务或者等待任务未来的执行结果。

public interface ExecutorService extends Executor {
 
    /**
     * 关闭线程池，调用此方法后不再接受新任务，但会处理线程池内
     * 尚未完成的任务，如果线程池已经关闭，再次调用此方法将无事发生。
     * 这个方法不会等待已提交但尚未完成的任务执行完毕，需要调用awaitTermination(long timeout, TimeUnit unit)
     * 来等待。
     */
    void shutdown();
 
    /**
     * 调用此方法会尝试停止所有正在运行的线程，比如：调用Thread.interrupt()
     * 标记线程已中断，如果任务没有响应中断则线程无法停止。这个方法会返回尚未
     * 执行的任务列表，它不会等待正在执行的任务执行完毕，需要调用awaitTermination(long timeout, TimeUnit unit)
     * 来等待。
     */
    List<Runnable> shutdownNow();
 
    /**
     * 判断线程池是否已关闭，true为关闭。
     */
    boolean isShutdown();
 
    /**
     * 如果调用shutdown()或者shutdownNow()后，所有任务都已完成，则返回true
     */
    boolean isTerminated();
 
    /**
     * 使当前调用此方法线程陷入阻塞，直到：
     *      1.调用停止线程池方法后，完成所有任务。
     *      2.阻塞超时。
     *      3.调用此方法线程被中断。
     *
     * @param timeout 最大等待时长
     * @param unit    时长单位
     * @return 如果任务都执行完毕返回true，如果超时或中断则返回false
     */
    boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException;
 
    /**
     * 提交一个有返回值的任务后，将返回一个Future对象代表任务的运行结果，
     * 可以调用Future.get()获得任务的执行结果，如果提交任务后想立即获得结果，
     * 可以用：result = exec.submit(aCallable).get();这样的方式获得，
     * 调用线程将陷入阻塞，直到线程池执行完任务，执行结果被放入到Future对象。
     *
     * @param task 待执行的任务。
     * @param <T>  任务执行结果的类型。
     * @return 返回Future对象，线程池执行完毕任务后，执行结果会被放入到Future对象。
     */
    <T> Future<T> submit(Callable<T> task);
 
    /**
     * 提交一个可执行的任务和给定的执行结果，并返回一个Future对象代表该任务将来的运行结果，
     * 如果任务成功执行，Future对象将返回我们给定的执行结果。
     *
     * @param task   待执行的任务
     * @param result 任务执行完毕后的返回结果
     * @param <T>    返回结果的类型
     * @return 返回Future对象，线程池执行完毕任务后，执行结果会被放入到Future对象。
     */
    <T> Future<T> submit(Runnable task, T result);
 
    /**
     * 提交待执行的任务并返回Future对象代表该任务未来的运行结果，如果任务执行成功，
     * 调用Future.get()将返回null。
     *
     * @param task 待执行的任务
     * @return 返回Future对象，代表该任务未来的运行结果。
     */
    Future<?> submit(Runnable task);
}

我们可以用ExecutorService来模拟一个网络服务，用Executors.newFixedThreadPool(int)工厂方法生成的线程池中的线程来处理传入的网络请求：

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

关闭ExecutorService分两个阶段，首先调用shutdown()拒绝再有新的任务提交，然后在必要的时候调用shutdownNow()，尝试中断正在执行的任务。

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

AbstractExecutorService

AbstractExecutorService是juc（java.util.concurrent）包下提供的ExecutorService接口的默认实现，在newTaskFor方法中返回FutureTask作为RunnableFuture接口的实现。可以看到不论是Runnable还是Callable类型的任务，都会被封装成RunnableFuture类型的任务来执行。

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);
    }
     
    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }
     
    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }
 
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }
     
}

ThreadPoolExecutor

ThreadPoolExecutor是ExecutorService的一种实现，可以用若干已经池化的线程执行被提交的任务。使用线程池可以帮助我们限定和整合程序资源，尽可能避免创建新的线程来执行任务从而降低任务调用的开销，在执行大量异步任务的时候反而能获得更好的性能。此外，ThreadPoolExecutor还会维护一些统计信息，比如已完成的任务数量。

juc包的作者Doug Lea推荐程序员尽量使用更为便利的Executors类的工厂方法来配置线程池：

Executors.newCachedThreadPool()：创建一个线程池，可以根据线程池的需要来创建任务。这个线程池比较适用执行周期较短且量大的异步任务，在调用execute(...)方法时如果线程池中存在闲置的线程，将复用闲置线程，否则创建一个新线程执行任务。如果线程的闲置时间超过60s，线程将被终止并从线程池内移除。因此，该线程池即便空闲时间再长，也不会有资源的消耗。
Executors.newFixedThreadPool(int nThreads)：创建一个线程池，nThreads为池内线程的数量，池内最多同时有nThreads个线程并行处理任务，如果有新的任务提交到线程池，会先暂存在线程池中的无边界任务队列进行等待，直到有线程可用。如果有线程在执行期间因为错误提前终止，线程池将启动一个新的线程代替原先的线程继续处理任务队列中处于等待的任务。除非显式调用shutdown()，否则线程池中的线程将一直存在。
Executors.newSingleThreadExecutor()：创建一个Executor，该Executor使用一个工作线程处理任务。如果线程在执行期间因为错误而终止，将启动一个新的线程代替原先的线程继续处理无边界任务队列中处于等待的任务。队列中的任务是按顺序执行，任何时刻都不会有多个任务处于活跃状态。与newFixedThreadPool(1)不同，newFixedThreadPool(int nThreads)生成的线程池，可以强转为ThreadPoolExecutor类型，再调用setCorePoolSize(int corePoolSize)方法设置核心线程数，而newSingleThreadExecutor()的实现类为FinalizableDelegatedExecutorService，无法直接设置核心线程数。

上面三种是较为常见的配置线程池的工厂方法，如果有需要根据业务场景特殊配置线程池的，请看下面的参数：

核心线程数和最大线程数

ThreadPoolExecutor将根据corePoolSize（核心线程数）和maximumPoolSize（最大线程数）设置的边界自动调整线程池内工作线程的数量（通过getPoolSize()），corePoolSize可以通过getCorePoolSize()、setCorePoolSize(int corePoolSize)获取和设置核心线程数，maximumPoolSize可以通过getMaximumPoolSize()、setMaximumPoolSize(int maximumPoolSize)获取和设置最大线程数。当有新的任务提交时，如果工作线程少于核心线程数，将会创建一个新线程来执行该任务，即便其他工作线程处于闲置状态。如果工作线程多于corePoolSize但少于maximumPoolSize，则当任务队列满的时候才会创建新线程。如果corePoolSize和maximumPoolSize数值一样，则创建一个固定大小的线程池；如果将maximumPoolSize设置为Integer.MAX_VALUE，则线程池可以容纳任意数量的并发任务。

按需构造

核心线程只有当有新任务到达时才会创建，但我们可以重写prestartCoreThread() 或者prestartAllCoreThreads()来预先启动核心线程。如果在构造一个线程池时，传入的任务队列已经存在任务，则需要线程池初始化完毕后，预先启动线程。

创建新线程

使用ThreadFactory（线程工厂）创建新线程。如果没有特别指定，则使用Executors.defaultThreadFactory()作为默认的线程工厂，该线程工厂所创建的线程都位于相同的线程组（ThreadGroup）中，线程的优先级都是NORM_PRIORITY，线程守护状态都为false。通过提供不同线程工厂的实现，你可以修改线程名、线程组、线程优先级和守护状态等等。

活跃时间

如果线程池中的线程数超过核心线程数，多出的线程如果空闲时间超出keepAliveTime（活跃时间）将会终止，回收不再活跃的线程。当有需要时，新的线程会重新创建。可以通过setKeepAliveTime(long time, TimeUnit unit)动态设置活跃时间。如果time设置为Long.MAX_VALUE，unit设置为TimeUnit.NANOSECONDS，那么多余的空闲线程将不会在关闭线程池之前回收。如果调用allowCoreThreadTimeOut(boolean value)传入的value为true，那么keepAliveTime将适用于核心线程，如果allowCoreThreadTimeOut为true且keepAliveTime不为0，核心线程的空闲时间超出活跃时间，核心线程也会被回收。

队列

阻塞队列（BlockingQueue）允许在获取元素时陷入等待，直到有元素加入到队列中。调用阻塞队列方法时，有些方法不一定马上返回，可能会在未来某个时刻达成某些条件时返回。阻塞队列的方法伴随四种形式：

抛出异常。
返回特殊值，null或者false，具体视操作而定。
调用线程无限期陷入阻塞直到某些条件达成。
限定阻塞时长。

	抛异常	特殊值	阻塞	超时
插入	add(e)	offer(e)	put(e)	offer(e, time, unit)
移除	remove()	poll()	take()	poll(time, unit)
检查（获取但不移除队列头部元素）	element()	peek()

阻塞队列不接受null元素，如果调用add、put、offer尝试添加一个null元素，将会抛出NullPointerException异常，当调用poll操作失败时也会返回null。阻塞队列可能有容量限制，无论何时都不能向队列添加超过剩余容量的元素，否则只能调用put方法陷入阻塞，直到有剩余的空间可以容纳元素。如果对队列的容纳空间没有限制，则剩余容量返回Integer.MAX_VALUE。阻塞队列的实现一般用于生产者-消费者队列的场景，此外阻塞队列还实现了Collection接口，因此，队列还可以使用remove(x)来移除元素。

阻塞队列是线程安全的，所有排队方法的实现都是用内部锁或者其他并发控制手段来实现原子性的。然而，除非是特殊规定，否则大部分集合操作，如：addAll、containsAll、retainAll 、removeAll不一定要保证原子性。因此，可能出现在调用addAll(c)时，只添加c中一部分的元素就抛出异常。阻塞队列本质上并不支持关闭的操作，如：close或shutdown，当有需要让队列不再接受新元素。如果有这种需要或者特性更倾向于以来队列的实现。一种常见的策略是生产者往队列插入具有特殊标识的对象，当消费者使用对象时，会对特殊标识进行解释。

注意，阻塞队列允许多个生产者和消费者同时使用，如下：

class Producer implements Runnable {
  private final BlockingQueue queue;
  Producer(BlockingQueue q) { queue = q; }
  public void run() {
    try {
      while (true) { queue.put(produce()); }
    } catch (InterruptedException ex) { ... handle ...}
  }
  Object produce() { ... }
}
 
class Consumer implements Runnable {
  private final BlockingQueue queue;
  Consumer(BlockingQueue q) { queue = q; }
  public void run() {
    try {
      while (true) { consume(queue.take()); }
    } catch (InterruptedException ex) { ... handle ...}
  }
  void consume(Object x) { ... }
}
 
class Setup {
  void main() {
    BlockingQueue q = new SomeQueueImplementation();
    Producer p = new Producer(q);
    Consumer c1 = new Consumer(q);
    Consumer c2 = new Consumer(q);
    new Thread(p).start();
    new Thread(c1).start();
    new Thread(c2).start();
  }
}

public interface BlockingQueue<E> extends Queue<E> {
    /**
     * 在不超过队列容量的情况下插入一个元素将返回true，，如果队列没有多余的空间抛出
     * IllegalStateException异常，当使用容量受限的队列时最好使用offer。
     *
     * @param e 待添加进队列的元素
     * @return 返回true代表元素加入队列成功
     */
    boolean add(E e);
 
    /**
     * 相比add(E)如果队列满时插入元素不报错，只是返回false。
     *
     * @param e 待添加进队列的元素
     * @return 返回true代表元素加入队列成功，队列满时无法插入返回false
     */
    boolean offer(E e);
 
    /**
     * 将一个元素插入到队列，如果有必要会等待队列有多余空间可以插入。如果调用
     * put(E)的线程被中断，将抛出中断异常InterruptedException
     *
     * @param e 待添加进队列的元素
     */
    void put(E e) throws InterruptedException;
 
    /**
     * 相比offer(E)多了插入元素时陷入等待，如果等待期间队列依旧
     * 没有多余的空间容纳元素，则返回false，如果等待期间能插入则返回true。
     * 如果等待期间线程被中断，则抛出中断异常InterruptedException。
     *
     * @param e       待添加进队列的元素
     * @param timeout 等待时长
     * @param unit    等待时长单位
     */
    boolean offer(E e, long timeout, TimeUnit unit)
            throws InterruptedException;
 
    /**
     * 取出并删除队列的头部元素，如果队列为空，则会陷入等待，直到队列有新的元素加入，
     * 如果等待期间线程被中断，将抛出中断异常
     *
     * @return 对头元素
     */
    E take() throws InterruptedException;
 
    /**
     * 相比take()多了一个等待时长，如果队列本身有元素，或者队列原先有空但等待期间有元素
     * 加入则返回头部元素，否则队列为空且等待期间没有元素加入，则返回null。如果等待期间调用线程
     * 被中断，则抛出InterruptedException异常。
     */
    E poll(long timeout, TimeUnit unit)
            throws InterruptedException;
 
    /**
     * 返回队列理想状况下可无阻塞容纳元素的容量。注意：我们不能通过此方法判断元素是否插入成功，
     * 因为可能存在别的线程插入或删除队列中的元素。
     */
    int remainingCapacity();
 
    /**
     * 从队列中移除指定的元素，如果队列中存在一个或多个相同的元素，即：o.equal(e)，则删除并返回true。
     */
    boolean remove(Object o);
 
    /**
     * 如果队列存在一个或多个相同的元素，即：o.equal(e)，则返回true。
     */
    boolean contains(Object o);
 
    /**
     * 删除此队列中所有可用元素，并将它们移动到给定的集合c中。当我们把元素从原队列取出时添加到集合c时，
     * 可能出现异常导致元素既不在原队列，也不在集合中。队列同样实现了Collection接口，如果将原队列当做
     * 参数传入将抛出IllegalArgumentException异常
     * @param c 将队列元素传输到给定的集合c。
     * @return 加入到集合c中元素的数量。
     */
    int drainTo(Collection<? super E> c);
 
    /**
     * 最多将maxElements个元素从队列传输到给定集合，其他和drainTo(Collection<? super E> c)一样。
     */
    int drainTo(Collection<? super E> c, int maxElements);
}

任何阻塞队列都可以用来获取和保存任务，如何使用队列视当前线程池的大小而定：

如果工作线程的数量小于corePoolSize，那么Executor更倾向于添加新线程执行而非让任务排队。
如果工作线程的数量大于等于corePoolSize，那么Executor更倾向于把任务添加到队列等待执行，而非创建新线程。
如果队列已满，且线程池内的线程数小于maximumPoolSize，Executor会创建一个新线程来执行任务，否则任务会被拒绝。

通常有三种排队策略：

同步队列（SynchronousQueue）：如果有线程从同步队列获取任务，则移交给线程，否则持有任务，如果有新的任务尝试入队将返回失败，可以根据入队结果判断是否要构造一个新线程。同步队列可以避免当处理多个请求时内部依赖出现锁定，直接交接任务要求对maximumPoolSize这一参数不做限制，即maximumPoolSize为Integer.MAX_VALUE，避免线程池拒绝提交任务。但如果线程池处理任务的速度不够快，可能出现线程无限增长。
无界队列（LinkedBlockingQueue）：如果线程池核心工作线程都在执行任务时，新提交的任务将在队列中等待。如果任务提交速度过快而执行任务的速度又慢，将导致队列中的任务无限增长。
有界队列（ArrayBlockingQueue）：当与有限的maximumPoolSize配合使用时，有界队列可以防止资源耗尽，但如何设定有界队列的大小是一个很难的问题。如果队列过大而maximumPoolSize过小，可以减少CPU的使用、操作系统资源和上下文切换的开销，这有可能导致低吞吐量。如果队列过小而maximumPoolSize过大，这会使得CPU十分繁忙，甚至出现巨大的调度开销，同样也会降低吞吐量。

拒绝任务

如果线程池关闭后有新任务提交、或者在任务队列已满的情况下，线程池到达最大线程数且所有线程都在执行任务，将调用RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor)拒绝任务。默认提供四种预定义拒绝策略：

ThreadPoolExecutor.AbortPolicy：默认情况下使用的策略，抛出RejectedExecutionException异常。
ThreadPoolExecutor.CallerRunsPolicy：使用调用线程执行任务，这种机制可以降低任务提交的速度。
ThreadPoolExecutor.DiscardPolicy：将无法入队也不能执行的任务直接丢弃。
ThreadPoolExecutor.DiscardOldestPolicy：如果线程池未关闭，则获取并丢弃队列的头部元素，再尝试用线程池执行任务，这一策略可能会失败，如果失败则重复之前的步骤。

除了上述四种步骤，我们也可以自定义拒绝策略。

钩子函数

ThreadPoolExecutor提供了可重写函数beforeExecute(java.lang.Thread, java.lang.Runnable)、afterExecute(java.lang.Runnable, java.lang.Throwable)，分别允许我们在执行任务前和执行任务后做一些操作，这些方法可以控制执行环境，例如：初始化ThreadLocals、收集统计信息、添加日志等等。此外，也可以重写terminated()方法，当线程池完全终止后会调用此方法。

如果钩子函数或者回调函数抛出异常，工作线程可能会终止。

队列维护

ThreadPoolExecutor提供了getQueue()允许外部获取队列进行监控和调试，但不管出于什么目的尽量少使用此方法。此外ThreadPoolExecutor还提供了remove(java.lang.Runnable) 和purge()用于删除任务，purge()可以取消大量处于排队等待的任务。

销毁

当一个线程池不再有引用指向，且线程池内没有存活线程将会自动关闭。如果你期望一个未手动调用shutdown()方法的线程池会被回收，你要设置合理的线程存活时间（keep-alive times）、设置核心线程数为0，或者设置allowCoreThreadTimeOut为true，当核心线程空闲时间超过存活时间将被回收，当线程池没有引用指向，且无存活线程，就会被自动关闭并回收。

源码解析

ThreadPoolExecutor的ctl变量类型为AtomicInteger，这个数值有32位，包含两个部分：

runState（运行状态）：线程池是否处于运行中、是否已关闭等等。
workerCount（工作线程数）：线程池当前有多少个存活（忙碌或空闲）的线程。

为了将运行状态和工作线程数放在一个int字段，我们划分前3位存储运行状态，后29位存储存活线程数量（2^29）-1（约5亿）。未来有可能调整ctl为AtomicLong类型，这可能需要调整移位和掩码，但如果使用AtomicInteger，ThreadPoolExecutor的代码会更简单也更高效一些。

workerCount是线程池中还存活的线程数，该值有时候可能会短暂不同于池内实际的存活线程数。当需要增加工作线程时，会先用CAS的方式对workerCount+1，然后才向ThreadFactory申请创建一个线程。

runState为线程池提供了生命周期控制，有以下几种状态：

RUNNING：允许接受新任务和处理队列中任务。
SHUTDOWN：不接受新任务，但处理队列中任务。
STOP：不接受新任务，不处理队列中任务，同时尝试中断正在执行的任务。
TIDYING：所有任务都终止，workerCount为0，线程池状态过度到TIDYING，将要执行terminated()钩子函数。
TERMINATED：terminated()函数执行完毕。

下面，我们来看看线程池状态的转换：

RUNNING->SHUTDOWN：调用shutdown()。
(RUNNING or SHUTDOWN)->STOP：调用shutdownNow()。
SHUTDOWN->TIDYING：当线程池不再有存活线程且队列为空。
STOP->TIDYING：当线程池不再有存活线程。
TIDYING->TERMINATED：调用terminated()。

检测线程池的状态从SHUTDOWN过度到TIDYING并非易事，因为在SHUTDOWN状态下，队列可能从非空变为空，即仍然有存活的线程处理队列中的任务。只有workerCount为0且队列为空，才能结束线程池。

public class ThreadPoolExecutor extends AbstractExecutorService {
    /**
     * 类型为AtomicInteger的ctl可以保证线程安全，该数值分两个部分：
     * 前3位为runState代表线程池当前的状态：RUNNING~TERMINATED，
     * 后29位为workerCount代表线程池内存活线程数量。
     */
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    /**
     * Integer.SIZE=32，COUNT_BITS=32-3=29，用COUNT_BITS来划分
     * runState和workerCount。
     */
    private static final int COUNT_BITS = Integer.SIZE - 3;
    /**
     * 1 << COUNT_BITS = 0010 0000 0000 0000 0000 0000 0000 0000
     * COUNT_MASK = (1 << COUNT_BITS) - 1 = 0001 1111 1111 1111 1111 1111 1111 1111
     * ~COUNT_MASK = 1110 0000 0000 0000 0000 0000 0000 0000
     * COUNT_MASK可以帮助我们计算线程池当前的runState和workerCount。
     * 假设ctl的值为：0000 0000 0000 0000 0000 0000 0000 0011
     * 调用runStateOf(ctl.get())，将做ctl.get() & ~COUNT_MASK运算：
     * 0000 0000 0000 0000 0000 0000 0000 0011
     * & 1110 0000 0000 0000 0000 0000 0000 0000
     * = 0000 0000 0000 0000 0000 0000 0000 0000
     * 由此我们可以得到，线程池当前状态为0，即SHUTDOWN。
     * 调用workerCountOf(ctl.get())，将做ctl.get() & COUNT_MASK运算：
     * 0000 0000 0000 0000 0000 0000 0000 0011
     * & 0001 1111 1111 1111 1111 1111 1111 1111
     * = 0000 0000 0000 0000 0000 0000 0000 0011
     * 由此我们可以得到，线程池还有3个存活的线程。
     */
    private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
 
    /**
     * -1的二进制表示为：1111 1111 1111 1111 1111 1111 1111 1111，
     * 左移29位为：1110 0000 0000 0000 0000 0000 0000 0000
     */
    private static final int RUNNING = -1 << COUNT_BITS;
    /**
     * 0的二进制表示为：0000 0000 0000 0000 0000 0000 0000 0000，
     * 左移29位和原先没有变化。
     */
    private static final int SHUTDOWN = 0 << COUNT_BITS;
    /**
     * 0的二进制表示为：0000 0000 0000 0000 0000 0000 0000 0001，
     * 左移29位为：0010 0000 0000 0000 0000 0000 0000 0000
     */
    private static final int STOP = 1 << COUNT_BITS;
    /**
     * 2的二进制表示为：0000 0000 0000 0000 0000 0000 0000 0010，
     * 左移29位为：0100 0000 0000 0000 0000 0000 0000 0000
     */
    private static final int TIDYING = 2 << COUNT_BITS;
    /**
     * 3的二进制表示为：0000 0000 0000 0000 0000 0000 0000 0011，
     * 左移29位为：0110 0000 0000 0000 0000 0000 0000 0000
     */
    private static final int TERMINATED = 3 << COUNT_BITS;
 
    private static int runStateOf(int c) {
        return c & ~COUNT_MASK;
    }
 
    private static int workerCountOf(int c) {
        return c & COUNT_MASK;
    }
 
    /**
     * 根据runState和workerCount生成ctl，比如初始化线程池时，
     * ctl = new AtomicInteger(ctlOf(RUNNING, 0))，代表线程池
     * 的状态为RUNNING，存活线程数量为0。
     * @param rs 线程池状态
     * @param wc 存活线程数量
     * @return
     */
    private static int ctlOf(int rs, int wc) {
        return rs | wc;
    }
 
    /**
     * 通过位运算在进行一些状态的判断时，我们不需要解析ctl的运行状态，
     * 假设当前ctl：1110 0000 0000 0000 0000 0000 0000 0011，
     * 我们要判断线程池状态是否小于STOP，由于ctl开头为1110，以补码的
     * 方式来计算，ctl的值必然为负，STOP开头为0010，以补码方式计算为正数，
     * 所以ctl必然小于STOP。
     * ctl的布局还能保证workerCount永远不会为负数。
     */
    private static boolean runStateLessThan(int c, int s) {
        return c < s;
    }
 
    /**
     * 判断线程池至少处于某个状态，假设线程池现在队列为空且无任何存活线程，
     * 所以能保证线程池处于TIDYING状态，如果s我们传入STOP，TIDYING的开头
     * 为0100，STOP的开头为0010，TIDYING>STOP，所以我们能知道，线程池至少
     * 处于STOP以上包含STOP的状态。
     */
    private static boolean runStateAtLeast(int c, int s) {
        return c >= s;
    }
 
    //线程池是否处于RUNNING状态
    private static boolean isRunning(int c) {
        return c < SHUTDOWN;
    }
    //CAS增加worker数量
    private boolean compareAndIncrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect + 1);
    }
    //CAS减少worker数量
    private boolean compareAndDecrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect - 1);
    }
}

我们知道AbstractExecutorService.submit(...)方法最终会调用execute(Runnable command)方法，而AbstractExecutorService类中并没有实现execute(Runnable command)方法，它将execute(Runnable command)的实现交由子类。那么我们来看看ThreadPoolExecutor又是如何实现execute(Runnable command)方法呢？当一个任务提交到线程池，它的执行流程又是如何呢？来看下面的代码注释：

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();
    //如果工作线程的数量小于核心线程的数量，则尝试增加工作线程
    if (workerCountOf(c) < corePoolSize) {//<1>
        /*
         * 如果成功增加工作线程，工作线程会执行我们提交的任务，我们就可以安心退出，
         * 但线程池可以并发提交任务，可能存在在<1>处时工作线程小于核心线程数，
         * 执行<2>处的addWorker(Runnable firstTask, boolean core)时，
         * 其他线程先当前线程提交任务并增加工作线程，线程池内工作线程数超过核心线程数，
         * 当前线程增加工作线程失败，不能直接退出。
         * 注：addWorker(Runnable firstTask, boolean core)，core为true
         * 代表增加核心线程，会将任务作为firstTask传入；而false代表增加非核心线程，
         * 如果传入firstTask为null，则代表让工作线程去队列中拉取任务。
         */
        if (addWorker(command, true))//<2>
            return;
        //如果<2>处增加工作线程失败，则重新获取ctl的值。
        c = ctl.get();
    }
    //判断线程池是否处于运行中，且任务可以入队成功，如果两者成立，则进入<3>分支
    if (isRunning(c) && workQueue.offer(command)) {//<3>
        int recheck = ctl.get();
        /*
         * 重新获取ctl，因为可能在进入<3>分支的时候，线程池被关闭，
         * 所以要重新判断线程池状态，如果线程池不是处于运行状态，且
         * 任务成功被移除，则进入<4>分支，拒绝任务。
         */
        if (!isRunning(recheck) && remove(command))//<4>
            reject(command);
        /*
         * 为什么这里要判断工作线程数量是否为0？因为如果设置allowCoreThreadTimeOut
         * 为true的话，核心线程是可以为0的，可能代码执行到<3>处workQueue.offer(command)之前，
         * 即任务还未入队，工作线程数量已经为0了，所以这里要重新根据ctl判断工作线程是否为0，
         * 如果为0得再增加非核心线程去队列拉取并执行任务。
         */
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    /*
     * 如果没有进入<3>分支，而到达<5>分支，一般分两种情况：
     * 1.线程池被关闭，<3>处isRunning(c)为false，此时调用<5>处的
     * addWorker(...)必然返回false，然后执行拒绝策略。
     * 2.线程池处于运行状态，<3>处isRunning(c)为true，但队列已满
     * workQueue.offer(command)返回false，入队失败。
     * 这时候应该尝试创建非核心工作线程执行任务，如果工作线程数量没到达最大线程数，
     * 则创建线程并执行任务，如果工作线程到达最大线程数，则addWorker(...)返回
     * false，执行拒绝策略。
     */
    else if (!addWorker(command, false))//<5>
        reject(command);
}

从ThreadPoolExecutor.execute(Runnable command)的实现，我们可以知道addWorker(Runnable firstTask, boolean core)方法是至关重要的，它决定了是否将任务添加进线程池执行。下面，我们再来看看addWorker(Runnable firstTask, boolean core)方法：

/**
 * 此方法会根据线程池当前的运行状态、线程池所设定的边界（核心线程数和最大线程数）。
 * 如果线程池允许创建线程执行任务，则创建线程执行firstTask并相应调整工作线程的数量。
 * 如果线程池状态处于已停止（STOP）、关闭（SHUTDOWN）则会返回false。如果向线程工
 * 厂请求创建线程失败，也会返回false。线程创建失败分两种情况，一种是线程工厂返回null，
 * 或者执行Thread.start()时出现异常（通常为OOM异常）。
 *
 * @param firstTask：新线程首要执行任务，如果没有则传入null。当工作线程数少于核心    线程数，线程池总是创建一个新线程来执行firstTask。
 * @param core：根据core为true或者false，决定是以核心线程数或者最大线程数作为界限， 判断当前线程池的工作线程池是否小于界限，如果小于则允许创建线程。
 * @return 如果成功添加工作线程则返回true。
 */
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (int c = ctl.get(); ; ) {//<1>
        // Check if queue empty only if necessary.
        /*
         * 在这个地方addWorker(...)会返回false，即添加工作线程失败，
         * 我们来看看是什么情况下会进入这个分支：
         * runStateAtLeast(c, SHUTDOWN)代表线程池运行状态至少处于
         * SHUTDOWN，如果线程池还处于RUNNING运行状态，此方法不会立即
         * 返回失败。所以我们知道，要进入此分支，首要条件就是运行状态大于
         * 等于SHUTDOWN。
         * 之后如果runStateAtLeast(c, STOP)、firstTask != null、
         * workQueue.isEmpty())这三个条件其一为true，则添加线程失败。
         * 首先是runStateAtLeast(c, STOP)，如果线程池当前处于STOP
         * 状态，这时候既不接受新任务，也不处理队列里的任务，所以不管
         * firstTask是否为null，都返回false。
         * 如果runStateAtLeast(c, STOP)为false，那运行状态只能是
         * SHUTDOWN，SHUTDOWN状态下会处理队列里的任务，但不再接受新
         * 任务，所以firstTask不为null，也直接返回false。
         * 如果运行状态既处于SHUTDOWN、firstTask也会空，且任务队列也
         * 为空，则毫无必要增加工作线程，也直接返回false。
         * 所以总结一下有两种情况不会进入此分支：
         * 1.线程池处于RUNNING状态的时候。
         * 2.线程池处于SHUTDOWN，但firstTask为空队列不为空时。
         */
        if (runStateAtLeast(c, SHUTDOWN)
                && (runStateAtLeast(c, STOP)
                || firstTask != null
                || workQueue.isEmpty()))
            return false;
 
        for (; ; ) {//<2>
            //根据core判断工作线程的上限，如果大于上限则返回false。
            if (workerCountOf(c)
                    >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
                return false;
            /*
             * 如果用CAS的方式成功增加工作线程的数量，则用break retry的方式
             * 结束了retry对应的外层循环（即<1>处for循环），而不是break所在
             * 的本层循环（即<2>处循环），代码会从<3>处开始执行。
             */
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            /*
             * 如果上面用CAS的方式增加工作线程失败，则会重新判断线程池当前
             * 状态是否至少处于SHUTDOWN，如果线程池已关闭，代码会跳到retry
             * 处重新执行<1>处的for循环。如果线程池仍然处于RUNNING状态，则
             * 重复执行<2>处的循环。
             */
            if (runStateAtLeast(c, SHUTDOWN))
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }
    //<3>
    boolean workerStarted = false;//如果工作线程启动成功，则赋值为true
    boolean workerAdded = false;//如果工作线程添加成功则赋值为true
    Worker w = null;
    try {
        //创建一个Worker对象，Worker对象会向线程工厂申请创建一个线程
        w = new Worker(firstTask);
        final Thread t = w.thread;
        //如果线程工厂创建的Thread对象不为null，则进入此分支
        if (t != null) {
            /*
             * 这里用可重入锁锁住try模块代码，因为要将之前创建好的
             * w对象放进workers集合。
             * 注：重入锁ReentrantLock的概念笔者会在以后的文章里
             * 单独介绍，这里先简单理解，可重入锁就是禁止其他线程同时
             * 访问mainLock.lock()到mainLock.unlock()之间的代码,
             * 和synchronized有些类似。
             */
            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();
                /*
                 * 重新获取ctl，判断线程池当前是否处于运行状态，或小于STOP状态，
                 * 即线程池处于RUNNING或SHUTDOWN，如果处于RUNNING则直接进入分支，
                 * 如果处于SHUTDOWN且首要执行任务为空，代表可能要启动一个工作线程
                 * 来执行队列中的任务。
                 */
                if (isRunning(c) ||
                        (runStateLessThan(c, STOP) && firstTask == null)) {
                    /*
                     * 判断线程是否已经启动，如果使用的是Executors.DefaultThreadFactory
                     * 默认的线程工厂，正常来说创建出来的Thread对象都是线程未启动的，即：尚未
                     * 调用Thread.start()。但ThreadPoolExecutor允许我们传入定制化的线程
                     * 工厂，所以会存在线程工厂创建出Thread对象，但Thread对象已调用过start()
                     * 方法的可能。
                     */
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    //将创建好的worker添加进集合workers。
                    workers.add(w);
                    //更新历史上最大的工作线程数，即workers.size()。
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    //将worker添加进workers后，更新workerAdded的值为true。
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            //如果worker成功加入集合，则启动线程，并更新workerStarted为true。
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        /*
         * 如果worker没有启动，代表worker没有加入到workers集合，
         * 可能线程池状态>=STOP，则需要执行添加工作线程失败操作。
         */
        if (!workerStarted)
            addWorkerFailed(w);
    }
    //返回工作线程是否启动成功
    return workerStarted;
}

Java并发之ThreadPoolExecutor源码解析（三）

Worker

先前，笔者讲解到ThreadPoolExecutor.addWorker(Runnable firstTask, boolean core)，在这个方法中工作线程可能创建成功，也可能创建失败，具体视线程池的边界条件，以及当前内存情况而定。

那么，如果线程池当前的状态，是允许创建Worker对象的，那么创建Worker的内部流程又是怎样呢？线程池为何要使用Worker包装Thread来创建一个线程，为何不直接使用原生的Thread来创建线程？如果创建Worker的firstTask不为空，那么Worker理所当然应该优先执行firstTask任务，如果firstTask为空，那Worker又要如何获取任务来执行呢？我们还有一堆亟待解决的问题。

首先我们来解决前两个问题，Worker的创建流程，以及为什么不使用原生Thread代替Worker？首先，Doug Lea用Worker包装Thread，意味着Worker比Thread拥有更多的功能。例如：Worker会统计它所对应的线程执行了多少任务、通过Worker可以知道线程是否已启动、线程是否正在执行任务？而这些信息都是原生Thread所没有的，所以需要一个Worker类来扩展Thread。

在创建Worker时，会先设置其state的值为-1，代表Worker所对应的线程尚未启动，即还没有调用Worker.thread.start()，之后会进行firstTask的赋值，向线程工厂申请创建线程，创建完毕后，等待外部调用Worker.thread.start()启动一个线程执行Worker.run()方法。

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    /*
     * 当初始化一个Worker时，会向线程工厂申请创建一个Thread对象
     * 用来执行任务，为null代表线程工厂创建失败。
     */
    final Thread thread;
    //首要执行任务，该字段可能为null。
    Runnable firstTask;
    //thread已完成任务数。
    volatile long completedTasks;
 
    Worker(Runnable firstTask) {
        /*
         * state初始为-1，代表还未调用Worker.thread.start()，
         * Worker对应的线程尚未被创建，还不能中断。线程启动后，
         * 如果线程正在执行任务，state为1，如果线程启动后没在
         * 执行任务的状态则state为0。
         */
        setState(-1);//<1>
        this.firstTask = firstTask;
        /*
         * 创建Thread对象的时候，会把Worker对象本身传入，而Worker
         * 本身实现了Runnable接口，当调用thead.start()启动一个线程
         * 执行thread.run()时，会进而调用Worker.run()方法。
         */
        this.thread = getThreadFactory().newThread(this);
    }
 
 
    //Worker对象将任务的执行委托给ThreadPoolExecutor.runWorker(Worker w).
    public void run() {
        runWorker(this);
    }
     
    //判断Worker是否处于被某个线程持有状态。
    protected boolean isHeldExclusively() {
        return getState() != 0;//<2>
    }
 
    /*
     * 线程尝试持有worker对象，如果worker没有被某个线程
     * 持有，则state为0，则用CAS的方式将worker的state
     * 改为1，并设置exclusiveOwnerThread为当前线程。
     */
    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {//<3>
            setExclusiveOwnerThread(Thread.currentThread());//<4>
            return true;
        }
        return false;
    }
 
    /*
     * 线程释放worker对象，将state改为0，exclusiveOwnerThread
     * 改为null。
     */
    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }
 
    /*
     * 调用父类acquire(int arg)时，会进而调用到Worker本身实现的
     * tryAcquire(int unused)。
     */
    public void lock() {
        acquire(1);//<5>
    }
 
    public boolean tryLock() {
        return tryAcquire(1);
    }
 
    /*
     * 调用父类的release(int arg)时，会进而调用到Worker本身实现的
     * tryRelease(int unused)。
     */
    public void unlock() {
        release(1);//<6>
    }
 
    public boolean isLocked() {
        return isHeldExclusively();
    }
 
    /*
     * 尝试中断worker的对应线程，如果线程已经启动。创建
     * worker时，state为-1，直到调用worker.thread.start()
     * 后，worker的state为0，如果worker的state>=0，则尝试
     * 中断线程。
     */
    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

从上面的代码我们可以注意到，<1>、<2>、<3>、<4>、<5>、<6>处的方法并不是Worker本身有的方法，而是Worker继承自父类AbstractQueuedSynchronizer的方法。那么Worker为什么需要继承AbstractQueuedSynchronizer（AQS）？AQS又是何方神圣呢？

这里先简单介绍下AQS，它定义了若干接口交由程序员实现，诸如：lock()、unlock()、tryAcquire(int arg)、tryRelease(int arg)……等，以保证多个线程不会同时访问同一资源。ThreadPoolExecutor中的字段mainLock为可重入锁ReentrantLock，某种程度上来说也是实现了AQS，ThreadPoolExecutor通过mainLock的lock()、unlock()以保证线程池内一些非线程安全的对象不会出现并发读写，如：workers、completedTaskCount……等。

public class ReentrantLock implements Lock, java.io.Serializable {
    private final Sync sync;
     
    abstract static class Sync extends AbstractQueuedSynchronizer {//...}
     
    static final class NonfairSync extends Sync {//...}
     
    public ReentrantLock() {
        sync = new NonfairSync();
    }
    //...
}

那么Worker继承了AQS，Worker也实现了lock()、unlock()方法，说明Worker本身也存在多线程访问的可能，那是什么时候会出现多线程访问Worker呢？这里我们先按下这个问题，在介绍完Worker.run()之后你就会明白为何Worker要继承AQS以保证线程访问的顺序性。

我们知道当调用Worker.thread.start()方法时，会进而调用Worker.run()方法，而Worker.run()方法会进而调用ThreadPoolExecutor.runWorker(Worker w)。下面，我们来看看runWorker(Worker w)的执行流程：

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    /*
     * 将worker.firstTask设置为null，因为worker可能会存在较长的一段时间，
     * 而task可能很快执行完毕，避免worker长时间引用已完成task，以便GC回收
     * 已完成task。
     */
    w.firstTask = null;
    /*
     * 新创建的worker对象state为-1，调用worker.unlock()会进而调用
     * worker.tryRelease(int unused)，将state设置为0，代表worker
     * 对应工作线程已启动，线程处于可中断状态。
     */
    w.unlock(); // allow interrupts
    /*
     * 标志worker是否异常完成，如果在<1>处task为空，且无法通过
     * getTask()从任务队列中获取新的任务，则会跳出循环，并在<3>处
     * 赋值为false，代表worker并没有异常完成。
     * 不管worker是正常完成还是异常完成，最后都会将completedAbruptly的结果
     * 传给processWorkerExit(...)，如果是worker是正常退出，则将workerCount-1，
     * 并将worker从workers集合中移除。如果是异常退出，则不减少workerCount，仅仅
     * 是将异常worker从workers集合中移除，并尝试新增一个worker。
     */
    boolean completedAbruptly = true;
    try {
        /*
         * 如果task不为空，或者调用getTask()能任务队列中获取到新的任务，
         * 则进入while块的代码。如果任务队列中没有待执行的任务，调用getTask()
         * 会让当前线程陷入阻塞，直到超时或者有新任务进入任务队列。
         */
        while (task != null || (task = getTask()) != null) {//<1>
            /*
             * 如果能进入循环，代表worker准备开始执行任务，但在执行任务
             * 前会先上锁，等到任务执行结束又会在<2>处释放锁，然而线程池
             * 又不会让多个线程同时执行同一个任务，那么为什么在执行任务前
             * 要让worker先上锁，执行完毕再释放锁呢？
             * 我们假设有一个线程池有5个线程，其中A、B线程正在执行任务，
             * C、D、E处于空闲状态。所以我们能确定A、B两个worker已经
             * 获得了锁，而C、D、E还阻塞在getTask()方法中。现在线程池
             * 执行shutdown()方法，该方法会进而调用interruptIdleWorkers()
             * 中断处于空闲状态的工作线程。而在interruptIdleWorkers()方法中
             * 判断一个worker是否处于空闲，会调用worker.tryLock()，如果能
             * 成功获取到锁，则代表该worker处于空闲状态，则中断该worker对应的
             * 线程。因此，一个worker是有可能被多个线程访问的，比如worker本身
             * 对应的线程，又或者关闭线程池的线程。
             */
            w.lock();
            /*
             * 如果线程池的运行状态>=STOP，则中断当前线程。如果运行状态<STOP，
             * 则确保线程没有被中断。
             */
            if ((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 {
                /*
                 * 设置当前任务为空，这样就可以在下一次循环中获取新的任务。
                 * 对worker执行的任务数+1，并释放锁。
                 */
                task = null;
                w.completedTasks++;
                w.unlock();//<2>
            }
        }
        completedAbruptly = false;//<3>
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

在上面的runWorker(Worker w)中如果执行完worker的首要任务，或者首要任务为null，便会调用getTask()尝试从任务队列中获取任务，但调用getTask()可能会使当前线程陷入等待或者阻塞直到有任务入队。getTask()也有可能返回null导致当前worker对应的线程退出，有以下几个原因可能导致工作线程退出：

如果调用setMaximumPoolSize(int maximumPoolSize)改小最大线程数，导致工作线程数大于maximumPoolSize。
线程池运行状态为STOP。
线程池运行状态为SHUTDOWN，且队列为空。
线程等待任务超时后，如果线程池当前存在可回收的空闲线程（即allowCoreThreadTimeOut为true或者工作线程数大于核心线程数），如果队列为空，则可直接退出，如果队列不为空，工作线程数必须大于1，即线程池中最少两个工作线程，如果只有一个线程还退出的话，就会存在队列不为空，但线程池中没有一个工作线程的尴尬情况。

private Runnable getTask() {
    //超时标志，默认为false，获取任务如果超时则会在<5>赋值为true。
    boolean timedOut = false;
 
    for (; ; ) {
        int c = ctl.get();
 
        /*
         * 如果线程池处于RUNNING状态，则runStateAtLeast(c, SHUTDOWN)
         * 为false，不会再判断之后的逻辑为true或者false。
         * 如果线程池处于SHUTDOWN状态，且workQueue.isEmpty()为true，即
         * 任务队列为空，则直接返回。如果任务队列不为空，则无法进入if分支，
         * 依然要返回任务，按照SHUTDOWN的要求不再接受新任务，但仍要处理队列
         * 中的任务。
         * 如果线程池处于STOP状态，即便任务队列不为空，也不再处理，则直接进入
         * if分支后返回。
         * 所以总结一下，只有两种情况不会进入此分支：
         * 1.线程池处于RUNNING状态。
         * 2.线程池处于SHUTDOWN状态且任务队列不为空。
         */
        if (runStateAtLeast(c, SHUTDOWN)
                && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {//<1>
            decrementWorkerCount();
            return null;
        }
 
        int wc = workerCountOf(c);
 
        /*
         * timed决定是如果任务队列没有任务的话，是以无限期的方式
         * 等待任务入队，或者一旦等待时间超过keepAliveTime，则
         * 返回null。
         * 如果allowCoreThreadTimeOut为true，则核心线程等待
         * 任务时间超过keepAliveTime后会被回收。
         * 如果当前工作线程数量workerCount大于核心线程数，也会在
         * 线程等待任务超过keepAliveTime后回收线程。
         */
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        /*
         * 我们先看进入此分支后会做的事，再分析如何进入这个分支。进入
         * 此分支后，会用CAS减少workerCount的数量，成功则返回null。
         * 否则continue重新开始新一轮的for循环。
         * 现在，我们来分析下进入此分支的逻辑：
         * 首先是wc > maximumPoolSize,一般workerCount不会大于
         * maximumPoolSize，除非线程池运行期间通过
         * setMaximumPoolSize(int maximumPoolSize)将线程池
         * 的最大线程数改小。
         * (timed && timedOut)在第一轮for循环永远为false，因为
         * timedOut要为true的条件，首先是timed为true，即线程池内存在空闲后
         * 可回收的线程，不管是线程池允许回收核心线程，或者线程数大于核心线程数。
         * 只有在<4>获取任务超时后workQueue返回null，才有可能到达<5>处将
         * timedOut赋值为true，并且开始新一轮的循环。
         * 之后的两个判断wc>1和workQueue.isEmpty()，判断队列为空还好理解，
         * 为什么要判断wc>1?首先我们要知道maximumPoolSize必须大于等于1，当我们
         * 往线程池传入的maximumPoolSize<=0会抛出异常。其次，如果我们将<2>处改为：
         * (wc >= 1 || workQueue.isEmpty())，有可能出现线程池内只有一个线程，
         * 但任务队列不为null。依旧会进入此分支内部执行<3>的代码以CAS的方式对
         * workerCount-1，从而出现一个尴尬的情况，任务队列中有任务，但工作线程数
         * 为0。所以<2>处必须保证wc>1。
         * 思考一种情况：假设一个线程池核心线程数为3，最大线程数为5，
         * allowCoreThreadTimeOut为false，线程池当前工作线程数量也为5，5个线程
         * 同时完成任务，并执行getTask()获取任务，可想而知timed为true，因为工作
         * 线程数（5）大于核心线程数（3），5个工作线程都是调用<4>处workQueue.poll(...)
         * 等待任务，超时则返回null。如果超时时间到达，3个核心线程如何重新进入等待状态，
         * 剩余2个线程如何被回收？
         * 当5个线程超时返回后，会将timedOut赋值为true，然后重新开始新一轮的for循环，一直
         * 执行到此分支，此时(timed && timedOut)都为true，队列也都为null，所以5个线程会
         * 进入此分支。用CAS成功对workerCount-1的线程将被回收，失败的线程则continue又开始
         * 新一轮的for循环，直到wc<=corePoolSize，timed为false，最后剩余的工作线程数调用
         * workQueue.take()无限期地等待任务的到来，除非线程被中断。
         */
        if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {//<2>
            if (compareAndDecrementWorkerCount(c))//<3>
                return null;
            continue;
        }
 
        try {
            /*
             * workQueue.poll(...)和workQueue.take()都有可能使当前线程陷入
             * 等待，直到返回任务，只不过前者相比后者多了一个超时时间，到达超时时间
             * 如果有任务入队，则r不为null，直接返回任务。
             * 线程等待期间，如果线程被中断，则会抛出InterruptedException异常。
             * 一般关闭线程池时，会尝试中断空闲线程，而处于等待任务的空闲线程会跳到<6>处，
             * 重新开始新一轮的for循环，并且在<1>处判断线程池处于SHUTDOWN或者STOP，
             * 从而退出。
             */
            Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) ://<4>
                    workQueue.take();
            if (r != null)
                return r;
            /*
             * 只有等待任务超时控制流才会执行到这里，将超时标志赋值为true，
             * 重新开始新一轮的for循环。
             */
            timedOut = true;//<5>
        } catch (InterruptedException retry) {
            timedOut = false;//<6>
        }
    }
}

在runWorker(Worker w)中有两种方式可以进到下面的处理线程退出processWorkerExit(Worker w, boolean completedAbruptly)方法：

工作线程执行任务遇到异常，从而跳出while循环。
工作线程获取任务超时等待，正常退出循环。

上面两种方式传递给processWorkerExit(...)的completedAbruptly是不同的，第一个方式传入的completedAbruptly为true，第二个方式为false，虽然worker是同一个。那么当completedAbruptly为true或者false，processWorkerExit(...)的流程又是怎么走的呢？

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    /*
     * 从runWorker(Worker w)传递而来的变量，标志worker是否意外完成，
     * 当worker执行任务时抛出异常，该变量为true。如果是意外完成，则表明
     * workerCount尚未-1。如果worker获取任务超时从而要让线程被回收，在
     * getTask()方法中会对workerCount-1。
     */
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();
 
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        /*
         * 获取可重入锁后，将worker已完成的任务数加到线程池已完成任务数，
         * 并将worker从workers集合中移除。
         */
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    //尝试终止线程。
    tryTerminate();
 
    int c = ctl.get();
    //如果线程池运行状态处于RUNNING或SHUTDOWN，则进入此分支。
    if (runStateLessThan(c, STOP)) {
        //如果线程是正常退出，则进入此分支
        if (!completedAbruptly) {
            /*
             * min为线程池创建核心线程后，允许最小的核心线程数，如果
             * allowCoreThreadTimeOut为true则代表核心线程可以被回收，
             * 则min为0，否则min为核心线程数量。
             * 如果线程池允许回收线程，且队列不为空，则判断在移除当前worker
             * 后，线程池工作线程的数量是否还大于等于1，避免出现队列里有任务
             * 但没有线程执行的情况，如果工作线程数大于等于1，则退出线程线程。
             * 否则调用addWorker(Runnable firstTask, boolean core)
             * 尝试增加新的线程。
             */
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && !workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

最后，我们还需要介绍关闭线程池之后做的操作。关闭线程池会修改线程池运行状态，在advanceRunState(int targetState)会使用CAS自旋的方式，将线程池状态修改为SHUTDOWN。之后调用interruptIdleWorkers()中断空闲线程，这里我们看到中断的时候调用worker的tryLock()和unlock()。Worker之所以继承AQS就是为了方便区分哪些worker正在执行任务，哪些worker处于空闲中，以便在关闭线程池时中断所有空闲的worker。

/**
 * 调用此方法后不再接受新任务，但会执行现有队列任务。
 * 如果调用此方法前该方法已被调用，则不会有任何效果。
 * 该方法不会等待所有任务完成，需要调用：
 * awaitTermination(long timeout, TimeUnit unit)
 */
public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);//设置线程池状态为SHUTDOWN
        interruptIdleWorkers();//中断空闲线程
        onShutdown(); //钩子方法，按用户需要实现。
    } finally {
        mainLock.unlock();
    }
    /*
     * 尝试终止线程池，可能线程池有任务在执行，当前线程终止失败。
     * 但随着工作线程逐个退出，最后一个工作线程将成功终止线程池。
     */
    tryTerminate();
}
 
private void advanceRunState(int targetState) {
    for (; ; ) {
        int c = ctl.get();
        /*
         * 如果线程池运行状态>=targetState，则
         * runStateAtLeast(c, targetState)为true直接退出。
         * 否则调用ctlOf(int rs, int wc)，根据运行状态和工作
         * 线程数生成的值以CAS自旋的方式set进ctl。
         */
        if (runStateAtLeast(c, targetState) ||
                ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
            break;
    }
}
 
//调用shutdown()会进而调用此方法，中断空闲线程。
private void interruptIdleWorkers() {
    interruptIdleWorkers(false);
}
 
private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            /*
             * 如果线程未被中断，则尝试获取worker的锁，如果能
             * 成功获取，代表worker线程处于空闲中。
             */
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            //如果onlyOne为true代表最多只中断一个线程
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

在中断空闲线程后shutdown()还会调用tryTerminate()，如果查看tryTerminate()的引用，可以发现不单单shutdown()有调用，像addWorkerFailed(Worker w)、processWorkerExit(...)……都有调用。这个方法旨在终止线程池，如果当前有线程关闭了线程池，线程池如果没有存活线程，或者线程都处于空闲状态，自然而然执行尝试终止线程池方法；如果关闭线程池时，线程池仍然有线程处于执行任务状态，无法终止线程池，就要靠这些工作线程在退出时终止线程池。

final void tryTerminate() {
    for (; ; ) {
        int c = ctl.get();
        /*
         * 如果线程池运行状态处于RUNNING、TIDYING则退出，
         * 或者运行状态小于STOP（即处于RUNNING、SHUTDOWN）
         * 且队列不为空，则退出。
         * 总结一下，只有当运行状态处于STOP时或者状态处于SHUTDOWN
         * 但队列为空，才不会进此分支。
         */
        if (isRunning(c) ||
                runStateAtLeast(c, TIDYING) ||//<1>
                (runStateLessThan(c, STOP) && !workQueue.isEmpty()))
            return;
        //如果工作线程数不为0，则尝试最多中断一个空闲线程后退出。
        if (workerCountOf(c) != 0) {
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
        /*
         * 在processWorkerExit(...)方法中如果worker是
         * 异常退出会对workerCount-1，如果是正常退出，则
         * workerCount在getTask()中-1。之后在processWorkerExit(...)
         * 中移除worker。
         * 不论worker是正常退出还是异常退出，终归workerCount会慢慢回到0。
         * 而processWorkerExit(...)中在对workerCount-1后，还会调用
         * tryTerminate()。因此一个被关闭的线程池，它的最后一个线程，会
         * 执行到此处，处理终止线程池的工作。
         */
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            /*
             * 用CAS设置线程池运行状态为TIDYING，可能存在多个线程同时
             * 调用shutdown()后并依次执行到这一步（因为要获得可重入锁），
             * 但只有一个线程可以CAS成功，其他线程CAS失败后返回<1>处后
             * 判断运行状态>=TIDYING则退出。
             */
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    //空函数，具体由用户实现，主要用于终止线程池后的操作。
                    terminated();
                } finally {
                    //最后设置线程池的状态为TERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
    }
}

查看全文

相关阅读:
这是一棵树吗
感染者
es6 语法
css3 flex 详解,可以实现div内容水平垂直居中
移动端实现复制内容至剪贴板小例子
jq+mui 阻止事件冒泡
移动端H5 判断IOS还是Android 平台
移动端布局 rem,和px
关于H5移动端开发 iPhone X适配
H5 微信公众号监听返回事件

原文地址：https://www.cnblogs.com/xinxihua/p/14461338.html