Java 并发基础之 Java 线程池详解
我相信大家都看过很多的关于线程池的文章，基本上也是面试的时候必问的，如果你在看过很多文章以后，还是一知半解的，那希望这篇文章能让你真正的掌握好 Java 线程池。
线程池是非常重要的工具，如果你要成为一个好的工程师，还是得比较好地掌握这个知识，很多线上问题都是因为没有用好线程池导致的。即使你为了谋生，也要知道，这基本上是面试必问的题目，而且面试官很容易从被面试者的回答中捕捉到被面试者的技术水平。
本文略长，边看文章边翻源码（Java7 和 Java8 都一样），建议想好好看的读者抽出至少 30 分钟的整块时间来阅读。当然，如果读者仅为面试准备，可以直接滑到最后的总结部分。
总览
开篇来一些废话。下图是 java 线程池几个相关类的继承结构：
先简单说说这个继承结构，Executor 位于最顶层，也是最简单的，就一个 execute(Runnable runnable) 接口方法定义。
ExecutorService 也是接口，在 Executor 接口的基础上添加了很多的接口方法，所以一般来说我们会使用这个接口。
然后再下来一层是 AbstractExecutorService，从名字我们就知道，这是抽象类，这里实现了非常有用的一些方法供子类直接使用，之后我们再细说。
然后才到我们的重点部分 ThreadPoolExecutor 类，这个类提供了关于线程池所需的非常丰富的功能。
另外，我们还涉及到下图中的这些类：
同在并发包中的 Executors 类，类名中带字母 s，我们猜到这个是工具类，里面的方法都是静态方法，如以下我们最常用的用于生成 ThreadPoolExecutor 的实例的一些方法：
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue());
}
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue());
}
另外，由于线程池支持获取线程执行的结果，所以，引入了 Future 接口，RunnableFuture 继承自此接口，然后我们最需要关心的就是它的实现类 FutureTask。到这里，记住这个概念，在线程池的使用过程中，我们是往线程池提交任务（task），使用过线程池的都知道，我们提交的每个任务是实现了 Runnable 接口的，其实就是先将 Runnable 的任务包装成 FutureTask，然后再提交到线程池。这样，读者才能比较容易记住 FutureTask 这个类名：它首先是一个任务（Task），然后具有 Future 接口的语义，即可以在将来（Future）得到执行的结果。
当然，线程池中的 BlockingQueue 也是非常重要的概念，如果线程数达到 corePoolSize，我们的每个任务会提交到等待队列中，等待线程池中的线程来取任务并执行。这里的 BlockingQueue 通常我们使用其实现类 LinkedBlockingQueue、ArrayBlockingQueue 和 SynchronousQueue，每个实现类都有不同的特征，使用场景之后会慢慢分析。
想要详细了解各个
BlockingQueue的读者，可以参考我的前面的一篇对 BlockingQueue 的各个实现类进行详细分析的文章。
https://www.jianshu.com/p/4f66833cc432
以上就是本文要介绍的知识，废话不多说，开始进入正文。
Executor 接口
/*
* @since 1.5
* @author Doug Lea
*/
public interface Executor {
void execute(Runnable command);
}
我们可以看到 Executor 接口非常简单，就一个 void execute(Runnable command) 方法，代表提交一个任务。为了让大家理解 Java 线程池的整个设计方案，我会按照 Doug Lea 的设计思路来多说一些相关的东西。
我们经常这样启动一个线程：
new Thread(new Runnable(){
// do something
}).start();
用了线程池 Executor 后就可以像下面这么使用：
Executor executor = anExecutor;
executor.execute(new RunnableTask1());
executor.execute(new RunnableTask2());
如果我们希望线程池同步执行每一个任务，我们可以这么实现这个接口：
class DirectExecutor implements Executor {
public void execute(Runnable r) {
r.run();// 这里不是用的new Thread(r).start()，也就是说没有启动任何一个新的线程。
}
}
我们希望每个任务提交进来后，直接启动一个新的线程来执行这个任务，我们可以这么实现：
class ThreadPerTaskExecutor implements Executor {
public void execute(Runnable r) {
new Thread(r).start(); // 每个任务都用一个新的线程来执行
}
}
我们再来看下怎么组合两个 Executor 来使用，下面这个实现是将所有的任务都加到一个 queue 中，然后从 queue 中取任务，交给真正的执行器执行，这里采用 synchronized 进行并发控制：
class SerialExecutor implements Executor {
// 任务队列
final Queue tasks = new ArrayDeque();
// 这个才是真正的执行器
final Executor executor;
// 当前正在执行的任务
Runnable active;
// 初始化的时候，指定执行器
SerialExecutor(Executor executor) {
this.executor = executor;
}
// 添加任务到线程池: 将任务添加到任务队列，scheduleNext 触发执行器去任务队列取任务
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
executor.execute(active);
}
}
}
当然了，Executor 这个接口只有提交任务的功能，太简单了，我们想要更丰富的功能，比如我们想知道执行结果、我们想知道当前线程池有多少个线程活着、已经完成了多少任务等等，这些都是这个接口的不足的地方。接下来我们要介绍的是继承自 Executor 接口的 ExecutorService 接口，这个接口提供了比较丰富的功能，也是我们最常使用到的接口。
ExecutorService
一般我们定义一个线程池的时候，往往都是使用这个接口：
ExecutorService executor = Executors.newFixedThreadPool(args...);
ExecutorService executor = Executors.newCachedThreadPool(args...);
因为这个接口中定义的一系列方法大部分情况下已经可以满足我们的需要了。
那么我们简单初略地来看一下这个接口中都有哪些方法：
public interface ExecutorService extends Executor {
// 关闭线程池，已提交的任务继续执行，不接受继续提交新任务
void shutdown();
// 关闭线程池，尝试停止正在执行的所有任务，不接受继续提交新任务
// 它和前面的方法相比，加了一个单词“now”，区别在于它会去停止当前正在进行的任务
List shutdownNow();
// 线程池是否已关闭
boolean isShutdown();
// 如果调用了 shutdown() 或 shutdownNow() 方法后，所有任务结束了，那么返回true
// 这个方法必须在调用shutdown或shutdownNow方法之后调用才会返回true
boolean isTerminated();
// 等待所有任务完成，并设置超时时间
// 我们这么理解，实际应用中是，先调用 shutdown 或 shutdownNow，
// 然后再调这个方法等待所有的线程真正地完成，返回值意味着有没有超时
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
// 提交一个 Callable 任务
Future submit(Callable task);
// 提交一个 Runnable 任务，第二个参数将会放到 Future 中，作为返回值，
// 因为 Runnable 的 run 方法本身并不返回任何东西
Future submit(Runnable task, T result);
// 提交一个 Runnable 任务
Future submit(Runnable task);
// 执行所有任务，返回 Future 类型的一个 list
List> invokeAll(Collection> tasks)
throws InterruptedException;
// 也是执行所有任务，但是这里设置了超时时间
List> invokeAll(Collection> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;
// 只有其中的一个任务结束了，就可以返回，返回执行完的那个任务的结果
T invokeAny(Collection> tasks)
throws InterruptedException, ExecutionException;
// 同上一个方法，只有其中的一个任务结束了，就可以返回，返回执行完的那个任务的结果，
// 不过这个带超时，超过指定的时间，抛出 TimeoutException 异常
T invokeAny(Collection> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
这些方法都很好理解，一个简单的线程池主要就是这些功能，能提交任务，能获取结果，能关闭线程池，这也是为什么我们经常用这个接口的原因。
FutureTask
在继续往下层介绍 ExecutorService 的实现类之前，我们先来说说相关的类 FutureTask。
Future Runnable
/ /
/ /
RunnableFuture
|
|
FutureTask
FutureTask 通过 RunnableFuture 间接实现了 Runnable 接口，
所以每个 Runnable 通常都先包装成 FutureTask，
然后调用 executor.execute(Runnable command) 将其提交给线程池
我们知道，Runnable 的 void run() 方法是没有返回值的，所以，通常，如果我们需要的话，会在 submit 中指定第二个参数作为返回值：
Future submit(Runnable task, T result);
其实到时候会通过这两个参数，将其包装成 Callable。它和 Runnable 的区别在于 run() 没有返回值，而 Callable 的 call() 方法有返回值，同时，如果运行出现异常，call() 方法会抛出异常。
public interface Callable {
V call() throws Exception;
}
在这里，就不展开说 FutureTask 类了，因为本文篇幅本来就够大了，这里我们需要知道怎么用就行了。
下面，我们来看看 ExecutorService 的抽象实现 AbstractExecutorService 。
AbstractExecutorService
AbstractExecutorService 抽象类派生自 ExecutorService 接口，然后在其基础上实现了几个实用的方法，这些方法提供给子类进行调用。
这个抽象类实现了 invokeAny 方法和 invokeAll 方法，这里的两个 newTaskFor 方法也比较有用，用于将任务包装成 FutureTask。定义于最上层接口 Executor中的 void execute(Runnable command) 由于不需要获取结果，不会进行 FutureTask 的包装。
需要获取结果（FutureTask），用
submit方法，不需要获取结果，可以用execute方法。
下面，我将一行一行源码地来分析这个类，跟着源码来看看其实现吧：
invokeAny和invokeAll方法占了这整个类的绝大多数篇幅，读者可以选择适当跳过，因为它们可能在你的实践中使用的频次比较低，而且它们不带有承前启后的作用，不用担心会漏掉什么导致看不懂后面的代码。
public abstract class AbstractExecutorService implements ExecutorService {
// RunnableFuture 是用于获取执行结果的，我们常用它的子类 FutureTask
// 下面两个 newTaskFor 方法用于将我们的任务包装成 FutureTask 提交到线程池中执行
protected RunnableFuture newTaskFor(Runnable runnable, T value) {
return new FutureTask(runnable, value);
}
protected RunnableFuture newTaskFor(Callable callable) {
return new FutureTask(callable);
}
// 提交任务
public Future submit(Runnable task) {
if (task == null) throw new NullPointerException();
// 1/. 将任务包装成 FutureTask
RunnableFuture ftask = newTaskFor(task, null);
// 2/. 交给执行器执行，execute 方法由具体的子类来实现
// 前面也说了，FutureTask 间接实现了Runnable 接口。
execute(ftask);
return ftask;
}
public Future submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
// 1/. 将任务包装成 FutureTask
RunnableFuture ftask = newTaskFor(task, result);
// 2/. 交给执行器执行
execute(ftask);
return ftask;
}
public Future submit(Callable task) {
if (task == null) throw new NullPointerException();
// 1/. 将任务包装成 FutureTask
RunnableFuture ftask = newTaskFor(task);
// 2/. 交给执行器执行
execute(ftask);
return ftask;
}
// 此方法目的：将 tasks 集合中的任务提交到线程池执行，任意一个线程执行完后就可以结束了
// 第二个参数 timed 代表是否设置超时机制，超时时间为第三个参数，
// 如果 timed 为 true，同时超时了还没有一个线程返回结果，那么抛出 TimeoutException 异常
private T doInvokeAny(Collection> 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();
//
List> futures= new ArrayList>(ntasks);
// ExecutorCompletionService 不是一个真正的执行器，参数 this 才是真正的执行器
// 它对执行器进行了包装，每个任务结束后，将结果保存到内部的一个 completionQueue 队列中
// 这也是为什么这个类的名字里面有个 Completion 的原因吧。
ExecutorCompletionService ecs =
new ExecutorCompletionService(this);
try {
// 用于保存异常信息，此方法如果没有得到任何有效的结果，那么我们可以抛出最后得到的一个异常
ExecutionException ee = null;
long lastTime = timed ? System.nanoTime() : 0;
Iterator> it = tasks.iterator();
// 首先先提交一个任务，后面的任务到下面的 for 循环一个个提交
futures.add(ecs.submit(it.next()));
// 提交了一个任务，所以任务数量减 1
--ntasks;
// 正在执行的任务数(提交的时候 +1，任务结束的时候 -1)
int active = 1;
for (;;) {
// ecs 上面说了，其内部有一个 completionQueue 用于保存执行完成的结果
// BlockingQueue 的 poll 方法不阻塞，返回 null 代表队列为空
Future f = ecs.poll();
// 为 null，说明刚刚提交的第一个线程还没有执行完成
// 在前面先提交一个任务，加上这里做一次检查，也是为了提高性能
if (f == null) {
if (ntasks > 0) {
--ntasks;
futures.add(ecs.submit(it.next()));
++active;
}
// 这里是 else if，不是 if。这里说明，没有任务了，同时 active 为 0 说明
// 任务都执行完成了。其实我也没理解为什么这里做一次 break？
// 因为我认为 active 为 0 的情况，必然从下面的 f.get() 返回了
// 2018-02-23 感谢读者 newmicro 的 comment，
// 这里的 active == 0，说明所有的任务都执行失败，那么这里是 for 循环出口
else if (active == 0)
break;
// 这里也是 else if。这里说的是，没有任务了，但是设置了超时时间，这里检测是否超时
else if (timed) {
// 带等待的 poll 方法
f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
// 如果已经超时，抛出 TimeoutException 异常，这整个方法就结束了
if (f == null)
throw new TimeoutException();
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
}
// 这里是 else。说明，没有任务需要提交，但是池中的任务没有完成，还没有超时(如果设置了超时)
// take() 方法会阻塞，直到有元素返回，说明有任务结束了
else
f = ecs.take();
}
/*
* 我感觉上面这一段并不是很好理解，这里简单说下。
* 1/. 首先，这在一个 for 循环中，我们设想每一个任务都没那么快结束，
* 那么，每一次都会进到第一个分支，进行提交任务，直到将所有的任务都提交了
* 2/. 任务都提交完成后，如果设置了超时，那么 for 循环其实进入了“一直检测是否超时”
这件事情上
* 3/. 如果没有设置超时机制，那么不必要检测超时，那就会阻塞在 ecs.take() 方法上，
等待获取第一个执行结果
* 4/. 如果所有的任务都执行失败，也就是说 future 都返回了，
但是 f.get() 抛出异常，那么从 active == 0 分支出去(感谢 newmicro 提出)
// 当然，这个需要看下面的 if 分支。
*/
// 有任务结束了
if (f != null) {
--active;
try {
// 返回执行结果，如果有异常，都包装成 ExecutionException
return f.get();
} catch (ExecutionException eex) {
ee = eex;
} catch (RuntimeException rex) {
ee = new ExecutionException(rex);
}
}
}// 注意看 for 循环的范围，一直到这里
if (ee == null)
ee = new ExecutionException();
throw ee;
} finally {
// 方法退出之前，取消其他的任务
for (Future f : futures)
f.cancel(true);
}
}
public T invokeAny(Collection> tasks)
throws InterruptedException, ExecutionException {
try {
return doInvokeAny(tasks, false, 0);
} catch (TimeoutException cannotHappen) {
assert false;
return null;
}
}
public T invokeAny(Collection> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return doInvokeAny(tasks, true, unit.toNanos(timeout));
}
// 执行所有的任务，返回任务结果。
// 先不要看这个方法，我们先想想，其实我们自己提交任务到线程池，也是想要线程池执行所有的任务
// 只不过，我们是每次 submit 一个任务，这里以一个集合作为参数提交
public List> invokeAll(Collection> tasks)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
List> futures = new ArrayList>(tasks.size());
boolean done = false;
try {
// 这个很简单
for (Callable t : tasks) {
// 包装成 FutureTask
RunnableFuture f = newTaskFor(t);
futures.add(f);
// 提交任务
execute(f);
}
for (Future f : futures) {
if (!f.isDone()) {
try {
// 这是一个阻塞方法，直到获取到值，或抛出了异常
// 这里有个小细节，其实 get 方法签名上是会抛出 InterruptedException 的
// 可是这里没有进行处理，而是抛给外层去了。此异常发生于还没执行完的任务被取消了
f.get();
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
}
}
}
done = true;
// 这个方法返回，不像其他的场景，返回 List，其实执行结果还没出来
// 这个方法返回是真正的返回，任务都结束了
return futures;
} finally {
// 为什么要这个？就是上面说的有异常的情况
if (!done)
for (Future f : futures)
f.cancel(true);
}
}
// 带超时的 invokeAll，我们找不同吧
public List> invokeAll(Collection> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
if (tasks == null || unit == null)
throw new NullPointerException();
long nanos = unit.toNanos(timeout);
List> futures = new ArrayList>(tasks.size());
boolean done = false;
try {
for (Callable t : tasks)
futures.add(newTaskFor(t));
long lastTime = System.nanoTime();
Iterator> it = futures.iterator();
// 每提交一个任务，检测一次是否超时
while (it.hasNext()) {
execute((Runnable)(it.next()));
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
// 超时
if (nanos <= 0)
return futures;
}
for (Future f : futures) {
if (!f.isDone()) {
if (nanos <= 0)
return futures;
try {
// 调用带超时的 get 方法，这里的参数 nanos 是剩余的时间，
// 因为上面其实已经用掉了一些时间了
f.get(nanos, TimeUnit.NANOSECONDS);
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
} catch (TimeoutException toe) {
return futures;
}
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
}
}
done = true;
return futures;
} finally {
if (!done)
for (Future f : futures)
f.cancel(true);
}
}
}
到这里，我们发现，这个抽象类包装了一些基本的方法，可是像 submit、invokeAny、invokeAll 等方法，它们都没有真正开启线程来执行任务，它们都只是在方法内部调用了 execute 方法，所以最重要的 execute(Runnable runnable) 方法还没出现，需要等具体执行器来实现这个最重要的部分，这里我们要说的就是 ThreadPoolExecutor 类了。
ThreadPoolExecutor
ThreadPoolExecutor 是 JDK 中的线程池实现，这个类实现了一个线程池需要的各个方法，它实现了任务提交、线程管理、监控等等方法。
我们可以基于它来进行业务上的扩展，以实现我们需要的其他功能，比如实现定时任务的类 ScheduledThreadPoolExecutor 就继承自 ThreadPoolExecutor。当然，这不是本文关注的重点，下面，还是赶紧进行源码分析吧。
首先，我们来看看线程池实现中的几个概念和处理流程。
我们先回顾下提交任务的几个方法：
public Future submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public Future submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public Future submit(Callable task) {
if (task == null) throw new NullPointerException();
RunnableFuture ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
一个最基本的概念是，submit 方法中，参数是 Runnable 类型（也有Callable 类型），这个参数不是用于 new Thread(runnable).start() 中的，此处的这个参数不是用于启动线程的，这里指的是任务，任务要做的事情是 run() 方法里面定义的或 Callable 中的 call() 方法里面定义的。
初学者往往会搞混这个，因为 Runnable 总是在各个地方出现，经常把一个 Runnable 包到另一个 Runnable 中。请把它想象成有个 Task 接口，这个接口里面有一个 run() 方法。
我们回过神来继续往下看，我画了一个简单的示意图来描述线程池中的一些主要的构件：
当然，上图没有考虑队列是否有界，提交任务时队列满了怎么办？什么情况下会创建新的线程？提交任务时线程池满了怎么办？空闲线程怎么关掉？这些问题下面我们会一一解决。
我们经常会使用 Executors 这个工具类来快速构造一个线程池，对于初学者而言，这种工具类是很有用的，开发者不需要关注太多的细节，只要知道自己需要一个线程池，仅仅提供必需的参数就可以了，其他参数都采用作者提供的默认值。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue());
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue());
}
这里先不说有什么区别，它们最终都会导向这个构造方法：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue 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;
}
基本上，上面的构造方法中列出了我们最需要关心的几个属性了，下面逐个介绍下构造方法中出现的这几个属性：
-
corePoolSize
核心线程数，不要抠字眼，反正先记着有这么个属性就可以了。
-
maximumPoolSize
最大线程数，线程池允许创建的最大线程数。
-
workQueue
任务队列，BlockingQueue 接口的某个实现（常使用
ArrayBlockingQueue和LinkedBlockingQueue）。 -
keepAliveTime
空闲线程的保活时间，如果某线程的空闲时间超过这个值都没有任务给它做，那么可以被关闭了。注意这个值并不会对所有线程起作用，如果线程池中的线程数少于等于核心线程数 corePoolSize，那么这些线程不会因为空闲太长时间而被关闭，当然，也可以通过调用
allowCoreThreadTimeOut(true)使核心线程数内的线程也可以被回收。 -
threadFactory
用于生成线程，一般我们可以用默认的就可以了。通常，我们可以通过它将我们的线程的名字设置得比较可读一些，如 Message-Thread-1， Message-Thread-2 类似这样。
-
handler：
当线程池已经满了，但是又有新的任务提交的时候，该采取什么策略由这个来指定。有几种方式可供选择，像抛出异常、直接拒绝然后返回等，也可以自己实现相应的接口实现自己的逻辑，这个之后再说。
除了上面几个属性外，我们再看看其他重要的属性。
Doug Lea 采用一个 32 位的整数来存放线程池的状态和当前池中的线程数，其中高 3 位用于存放线程池状态，低 29 位表示线程数（即使只有 29 位，也已经不小了，大概 5 亿多，现在还没有哪个机器能起这么多线程的吧）。我们知道，java 语言在整数编码上是统一的，都是采用补码的形式，下面是简单的移位操作和布尔操作，都是挺简单的。
共有 0 条评论