线程池的源码分析及其思想
站长
· 阅读数 12
前言
线程池对于每一个程序员来说都是十分重要的，这也是面试基本要问到的一个点，很多线上的问题都是线程池没有使用好导致的，重点是要了解线程池的设计思想，这一点是十分重要的。
在开始讲述之前，还是要先看看Java有关线程池的几个类的继承结构，如下图所示：
从上图中可以看出，Executor是置于最顶层的，也是最短最简单的一个接口，里面只有一个execute(Runnable runnable)方法，从名字上也可以看出，这是一个执行任务的方法。接着往下看到ExecutorService，它同样的也是一个接口，只不过在Executor接口的基础上定义了更多的接口方法，这些方法更加便利我们去管理线程池，一般来说我们都会使用这个接口。
然后接着往下看，可以看到AbstractExecutorService类，这个类就不再是接口了，是一个抽象类，抽象类中往往帮我们写好了很多基础的方法，可以让子类去直接使用。
最后一个类，也就到了我们的ThreadPoolExecutor类，这个类可以说是囊括了我们使用线程池的所有基础方法，所以这个类是我们最常见到的一个线程池类了。
除了上面说的几个类，还包括下面几个类（这几个类可以理解为线程池中的任务、线程池中的任务队列等等）：
同时，我们可能还会使用到一个工具类Executors，里面的方法都是静态方法，来帮助我们生成最常用的ThreadPoolExecutor的实例的一些方法，代码如下所示：
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
......
从上图中还可以看出还有Future类，这个类看名字就能猜出来，这个类是用于获取线程执行结果的类，这里主要是关注它的实现类FutureTask。这里只需要记住以下几个点：
- 我们使用线程池的时候，是往线程池中提交任务，让线程池中的线程去执行我们的任务
- 我们往线程池中提交的任务往往是实现了
Runnable接口的 - 如果我们想要获取线程池执行任务的结果，那么我们就将实现了
Runnable的任务包装成FutureTask，然后再提交到线程池中
图中还有BlockingQueue类，这个类也是十分重要的概念，通过名字就可以看出来它是一个队列，如果线程数达到corePoolSize，那么我们提交的每个任务都会提交到等待队列中，等待线程池中的线程来领取并执行任务。而这里的等待队列，通常我们使用其实现类，LinkedBlockingQueue、ArrayBlockingQueue 和 SynchronousQueue。每个实现类都有不同的特征，这个后面会慢慢分析。
以上就是线程池的一些基础的知识，下面开始正式分析，看到这里应该要清楚以下几个知识点：
- 线程池的继承体系
- 线程池中的任务是什么
- 线程池中的等待队列
Executor接口
我们从线程池的顶层来依次往下说明，首先看到Executor的源码：
/*
* @since 1.5
* @author Doug Lea
*/
public interface Executor {
void execute(Runnable command);
}
这个接口是最简单的接口，只是定义了提交任务的方法 void execute(Runnable command)，注意这里的用词，是提交任务，并不是执行任务，这一点很重要（这也和实现相关，但是要记住这一步不一定就会直接执行任务）。
首先我们来简单回顾一下启动线程去执行任务：
new Thread(new Runnable(){
//......
}).start();
那么我们如果使用线程池，就会像如下这么去使用：
Executor executor = new Executor();
executor.execute(new RunnableTask()); // 这一步不一定会直接执行任务
上面就是线程池中去执行任务的伪代码了，可以看到这里只是将任务提交进去，什么时候执行是和线程池的实现有关的，大多数时候是不会立马执行相关任务，而是经过一系列的逻辑判断之后，再决定这个任务什么时候执行，是否应该执行等。
ExecutorService
这个接口的方法就十分丰富了，一般我们定义线程池往往也是使用这个接口，如下所示：
ExecutorService executor = Executors.newFixedThreadPool(args...);
这个接口内定义的方法大部分情况下都满足我们的使用条件了，源码如下所示：
public interface ExecutorService extends Executor {
// 关闭线程池 已提交的任务继续执行 不接受新提交的任务
void shutdown();
// 关闭线程池 尝试停止正在执行的所有任务同时不接受新提交的任务
// 可以看到它多了一个 now 所以它会立即去尝试停止当前正在执行的任务
List<Runnable> shutdownNow();
// 判断线程池是否已关闭
boolean isShutdown();
// 如果调用了 shutdown() 或 shutdownNow() 方法后 所有任务结束了 那么返回true
boolean isTerminated();
// 等待所有任务完成 并设置超时时间 返回值说明是否超时
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
// 提交一个 Callable 任务
<T> Future<T> submit(Callable<T> task);
// 提交一个 Runnable 任务 第二个参数将会作为返回值放到 Future 中
<T> Future<T> submit(Runnable task, T result);
// 提交一个 Runnable 任务
Future<?> submit(Runnable task);
// 执行所有任务 返回 Future 类型的一个 list
<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;
}
上述源码都有注释，一个基本的线程池的功能都包含在内了。
FutureTask
继续往下之前，我们要先了解FutureTask类（任务类）。
从图中可以看出FutureTask通过RunnableFuture接口，间接实现了Future类和Runnable类，所以每个Runnable通常都会包装成FutureTask，然后再提交到线程池中去。
那么使用过Runnable的都知道，它的run方法是不带有任何返回值的，所以如果我们需要返回值的话，通过上述的ExecutorService中可以看到会在submit方法中指定第二个参数作为返回值：
<T> Future<T> submit(Runnable task, T result);
到时候会通过这两个参数，包装成Callable，而Callable的call方法是有返回值的，如果运行异常，call方法也会抛出异常：
public interface Callable<V> {
V call() throws Exception;
}
为了方便大家的理解，下面放出一点FutrureTask类的源码：
......
private Callable<V> callable;
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
......
从源码中可以看出会将Runnable包装成Callable去用于获取任务的结果，这里我们只需要记住有这么一回事，是这么用的就可以了。
AbstractExecutorService
接下来就来到抽象类了，抽象类中往往会实现几个基础的方法，这些方法供子类去调用。
下面我们来看到这个抽象类的源码：
public abstract class AbstractExecutorService implements ExecutorService {
// RunnableFuture 用于获取执行任务的结果的
// 下面两个newTaskFor方法将我们的任务包装成FutureTask提交到线程池中执行
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();
// 1. 将任务包装成 FutureTask
RunnableFuture<Void> ftask = newTaskFor(task, null);
// 2. 交给执行器执行 execute 方法由具体的子类来实现
// 前面也说了 FutureTask 间接实现了Runnable 接口
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
// 1. 将任务包装成 FutureTask
RunnableFuture<T> ftask = newTaskFor(task, result);
// 2. 交给执行器执行
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
// 1. 将任务包装成 FutureTask
RunnableFuture<T> ftask = newTaskFor(task);
// 2. 交给执行器执行
execute(ftask);
return ftask;
}
// 提交任务集合tasks 只要有一个任务完成就可以结束
// 参数timed 代表是否设置超时机制 参数 nanos 代表超时时间
// 如果超时时间到了还没有任务执行结束则抛出TimeoutException异常
private <T> T doInvokeAny(Collection<? extends Callable<T>> 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<Future<T>> futures= new ArrayList<Future<T>>(ntasks);
// ExecutorCompletionService 不是真正的执行器 参数 this 才是真正的执行器
// 它对执行器进行了包装 每个任务结束后 将结果保存到内部的一个 completionQueue 队列中
ExecutorCompletionService<T> ecs =
new ExecutorCompletionService<T>(this);
try {
// 用于保存异常信息 此方法如果没有得到任何有效的结果 那么我们可以抛出最后得到的一个异常
ExecutionException ee = null;
long lastTime = timed ? System.nanoTime() : 0;
Iterator<? extends Callable<T>> it = tasks.iterator();
// 首先先提交一个任务 后面的任务到下面的 for 循环一个个提交
futures.add(ecs.submit(it.next()));
// 提交了一个任务 任务数量减 1
--ntasks;
// 正在执行的任务数
int active = 1;
for (;;) {
// ecs 上面说了 其内部有一个 completionQueue 用于保存执行完成的结果
// BlockingQueue 的 poll 方法不阻塞 返回 null 代表队列为空
Future<T> f = ecs.poll();
// 为 null 说明刚刚提交的第一个线程还没有执行完成
// 在前面先提交一个任务 加上这里做一次检查 也是为了提高性能
if (f == null) {
if (ntasks > 0) {
--ntasks;
futures.add(ecs.submit(it.next()));
++active;
}
// 这里是 else if 不是 if 这里说明 没有任务了 同时 active 为 0
// 这里的 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. 任务如果都提交了 如果设置了超时 那么就会一直循环检测是否超时
* 3. 如果没有设置超时机制 那么就会一直阻塞在 ecs.take()方法上 等待获取第一个执行结果
* 4. 如果所有的任务都执行失败 也就是说 future 都返回了
但是 f.get() 抛出异常 那么从 active == 0 分支出去
// 当然，这个需要看下面的 if 分支。
*/
// 有任务结束了 此时还在for循环内
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<T> f : futures)
f.cancel(true);
}
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
try {
return doInvokeAny(tasks, false, 0);
} catch (TimeoutException cannotHappen) {
assert false;
return null;
}
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return doInvokeAny(tasks, true, unit.toNanos(timeout));
}
// 执行所有的任务 返回任务结果
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
boolean done = false;
try {
// 这个很简单
for (Callable<T> t : tasks) {
// 包装成 FutureTask
RunnableFuture<T> f = newTaskFor(t);
futures.add(f);
// 提交任务
execute(f);
}
for (Future<T> f : futures) {
if (!f.isDone()) {
try {
// 这是一个阻塞方法 直到获取到值 或抛出了异常
// 这里有个小细节 其实 get 方法签名上是会抛出 InterruptedException 的
// 可是这里没有进行处理 而是抛给外层去了 此异常发生于还没执行完的任务被取消了
f.get();
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
}
}
}
done = true;
// 这个方法返回是真正的返回 任务都结束了
return futures;
} finally {
// 上面说的有异常的情况
if (!done)
for (Future<T> f : futures)
f.cancel(true);
}
}
// 带超时的 invokeAll
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
if (tasks == null || unit == null)
throw new NullPointerException();
long nanos = unit.toNanos(timeout);
List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
boolean done = false;
try {
for (Callable<T> t : tasks)
futures.add(newTaskFor(t));
long lastTime = System.nanoTime();
Iterator<Future<T>> 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<T> 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<T> f : futures)
f.cancel(true);
}
}
}
通过上面的源码可以看到抽象类实现了一些基本的方法，但是都没有真正的去开启线程去执行任务，而是内部抛给执行器调用了execute方法，所以接下来就要说到ThreadPoolExecutor类了。
ThreadPoolExecutor
这个类实现了一个线程池需要的各个方法，可以基于它来进行业务上的拓展，在介绍它之前，我们先来回顾一下线程池中提交任务的几个方法：
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;
}
这里需要提到一个概念，就是这里的参数Runnable task不是用于启动线程的那个Runnable，这里指的是任务的意思，任务要做的事情就是task中的run方法里定义的事情，或者Callable中call方法定义的事情。
那么接下来就开始分析ThreadPoolExecutor这个类了，我们往往会使用如下构造方法去构造一个线程池：
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.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：用于生成线程，一般使用默认
- handler：当线程池已经满了，但是又有新的任务提交的时候，该采取什么策略由这个来指定（拒绝策略）
除了上述的几个属性，还有一些其他比较重要的属性如下，线程池采用一个 32 位的整数来存放线程池的状态和当前池中的线程数，其中高 3 位用于存放线程池状态，低 29 位表示线程数（大概5亿多）。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 这里 COUNT_BITS 设置为 29(32-3) 意味着前三位用于存放线程状态 后29位用于存放线程数
private static final int COUNT_BITS = Integer.SIZE - 3;
// 000 11111111111111111111111111111
// 这里得到的是 29 个 1 也就是说线程池的最大线程数是 2^29-1=536870911
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 线程池的状态存放在高 3 位中
// 运算结果为 111跟29个0：111 00000000000000000000000000000
private static final int RUNNING = -1 << COUNT_BITS;
// 000 00000000000000000000000000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 001 00000000000000000000000000000
private static final int STOP = 1 << COUNT_BITS;
// 010 00000000000000000000000000000
private static final int TIDYING = 2 << COUNT_BITS;
// 011 00000000000000000000000000000
private static final int TERMINATED = 3 << COUNT_BITS;
// 将整数 c 的低 29 位修改为 0 就得到了线程池的状态
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 将整数 c 的高 3 为修改为 0 就得到了线程池中的线程数
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
/*
* Bit field accessors that don't require unpacking ctl.
* These depend on the bit layout and on workerCount being never negative.
*/
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：不接受新的任务，但是会继续处理等待队列中的任务
- STOP：不接受新的任务，不处理等待队列中的任务，中断正在执行任务的线程
- TIDYING：所有的任务都销毁了，workCount 为 0，线程池的状态在转换为 TIDYING 状态时，会执行钩子方法 terminated()
- TERMINATED：terminated() 方法结束后，线程池的状态就会变成这个
下面将介绍一下各个状态之间的转换：
- RUNNING -> SHUTDOWN：当调用了 shutdown() 后，会发生这个状态转换
- (RUNNING or SHUTDOWN) -> STOP：当调用 shutdownNow() 后，会发生这个状态转换
- SHUTDOWN -> TIDYING：当任务队列和线程池都清空后，会由 SHUTDOWN 转换为 TIDYING
- STOP -> TIDYING：当任务队列清空后，发生这个转换
- TIDYING -> TERMINATED：当 terminated() 方法结束后
接下来我们还需要看看一个内部类Worker(工人)，它就是线程池中做任务的线程，由它去执行相应的任务。
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
// 这个是真正执行任务的线程
final Thread thread;
// 如果这个线程起来以后 需要执行的第一个任务就可以放在这里 同样也可以不执行 线程可以自己到任务队列中去获取任务
Runnable firstTask;
// 用于存放此线程完成的任务数
volatile long completedTasks;
// Worker 只有这一个构造方法 传入 firstTask 也可以传 null
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 调用 ThreadFactory 来创建一个新的线程
this.thread = getThreadFactory().newThread(this);
}
// 这里调用了外部类的 runWorker 方法
public void run() {
runWorker(this);
}
...// 其他几个方法就是用 AQS 操作 来获取这个线程的执行权 用了独占锁
}
那么接下来就到了重点方法execute了，源码如下所示：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// 获取线程状态和线程数的那个整数
int c = ctl.get();
// 如果当前线程数少于核心线程数 那么直接添加一个 worker 来执行任务
// 创建一个新的线程 并把当前任务 command 作为这个线程的第一个任务(firstTask)
if (workerCountOf(c) < corePoolSize) {
// 添加任务成功 代表线程池接收了这个任务 就可以结束了
// 至于执行的结果 到时候会包装到 FutureTask 中
// 返回 false 代表线程池不允许提交任务
if (addWorker(command, true))
return;
c = ctl.get();
}
// 能到这里说明 1 当前线程数大于等于核心线程数 2 刚刚 addWorker 失败了
// 如果线程池处于 RUNNING 状态 把这个任务添加到任务队列 workQueue 中
if (isRunning(c) && workQueue.offer(command)) {
/* 这里面说的是 如果任务进入了 workQueue 判断我们是否需要开启新的线程
* 因为线程数在 [0, corePoolSize) 是无条件开启新的线程
* 如果线程数已经大于等于 corePoolSize 那么将任务添加到队列中 然后进到这里
*/
int recheck = ctl.get();
// 如果线程池已不处于 RUNNING 状态 那么移除已经入队的这个任务 并且执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
// 如果线程池还是 RUNNING 的 并且线程数为 0 那么开启新的线程
// 到这里 我们知道了 这块代码的真正意图是：担心任务提交到队列中了 但是线程都关闭了
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果 workQueue 队列满了 那么进入到这个分支
// 以 maximumPoolSize 为界创建新的 worker，
// 如果失败 说明当前线程数已经达到 maximumPoolSize 执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}
通过上面的代码可以看出线程池创建线程的步骤：
- 如果线程数在[0, corePoolSize) 是无条件开启新的线程
- 如果线程数在大于corePoolSize，那么将任务添加到队列中
- 如果队列满了，那么以maximumPoolSize为界创建新的线程去执行任务
那么接下来我们来看到它是怎么创建新的线程的，也就是看到addWorker(Runnable firstTask, boolean core) 方法：
// 第一个参数是线程起来以后该线程第一个执行的任务 可以为null
// 第二个参数是表示使用corePoolSize作为创建线程的界限（true） 还是使用maximumPoolSize作为界限（false）
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 如果线程池已关闭 并满足以下条件之一 那么不创建新的 worker：
// 1. 线程池状态大于 SHUTDOWN 其实也就是 STOP、TIDYING、或 TERMINATED
// 2. firstTask != null
// 3. workQueue.isEmpty()
// 简单分析下：
// 状态控制的问题 当线程池处于 SHUTDOWN 的时候 不允许提交任务 但是已有的任务继续执行
// 当状态大于 SHUTDOWN 时 不允许提交任务 且中断正在执行的任务
// 如果线程池处于 SHUTDOWN 但是 firstTask 为 null 且 workQueue 非空 那么是允许创建 worker 的
// 这是因为 SHUTDOWN 的语义：不允许提交新的任务 但是要把已经进入到 workQueue 的任务执行完 所以在满足条件的基础上 是允许创建新的 Worker 的
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;
// 由于有并发 重新再读取一下 ctl
c = ctl.get();
// 正常如果是 CAS 失败的话 进到下一个里层的for循环就可以了
// 可是如果是因为其他线程的操作 导致线程池的状态发生了变更 如有其他线程关闭了这个线程池
// 那么需要回到外层的for循环
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
// 到这里 说明可以开始创建线程来执行任务了
// worker 是否已经启动
boolean workerStarted = false;
// 是否已将这个 worker 添加到 workers 这个 HashSet 中
boolean workerAdded = false;
Worker w = null;
try {
final ReentrantLock mainLock = this.mainLock;
// 把 firstTask 传给 worker 的构造方法
w = new Worker(firstTask);
// 取 worker 中的线程对象，之前说了，Worker的构造方法会调用 ThreadFactory 来创建一个新的线程
final Thread t = w.thread;
if (t != null) {
// 这个是整个线程池的全局锁 持有这个锁才能让下面的操作“顺理成章”，
// 因为关闭一个线程池需要这个锁 至少我持有锁的期间 线程池不会被关闭
mainLock.lock();
try {
int c = ctl.get();
int rs = runStateOf(c);
// 小于 SHUTTDOWN 那就是 RUNNING 是最正常的情况
// 如果等于 SHUTDOWN 则不接受新的任务 但是会继续执行等待队列中的任务
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// worker 里面的 thread 可不能是已经启动的
if (t.isAlive())
throw new IllegalThreadStateException();
// 加到 workers 这个 HashSet 中
workers.add(w);
int s = workers.size();
// largestPoolSize 用于记录 workers 中的个数的最大值
// 因为 workers 是不断增加减少的 通过这个值可以知道线程池的大小曾经达到的最大值
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 添加成功的话 启动这个线程
if (workerAdded) {
// 启动线程
t.start();
workerStarted = true;
}
}
} finally {
// 如果线程没有启动 需要做一些清理工作
if (! workerStarted)
addWorkerFailed(w);
}
// 返回线程是否启动成功
return workerStarted;
}
继续往下走，worker中线程start后，run方法会调用runWorker方法：
// Worker 类的 run() 方法
public void run() {
runWorker(this);
}
继续：
// 此方法由 worker 线程启动后调用 这里用一个 while 循环来不断地从等待队列中获取任务并执行
final void runWorker(Worker w) {
//
Thread wt = Thread.currentThread();
// 该线程的第一个任务(如果有的话)
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 循环调用 getTask 获取任务
while (task != null || (task = getTask()) != null) {
w.lock();
// 如果线程池状态大于等于 STOP 那么意味着该线程也要中断
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) {
// 这里不允许抛出 Throwable 所以转换为 Error
thrown = x; throw new Error(x);
} finally {
// 也是一个钩子方法 将 task 和异常作为参数 留给需要的子类实现
afterExecute(task, thrown);
}
} finally {
// 置空 task 准备 getTask 获取下一个任务
task = null;
// 累加完成的任务数
w.completedTasks++;
// 释放掉 worker 的独占锁
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 如果到这里 需要执行线程关闭：
// 1. 说明 getTask 返回 null 也就是说 队列中已经没有任务需要执行了 执行关闭
// 2. 任务执行过程中发生了异常
// 第一种情况 已经在代码处理了将 workCount 减 1 这个在 getTask 方法分析中会说
// 第二种情况 workCount 没有进行处理 所以需要在 processWorkerExit 中处理
processWorkerExit(w, completedAbruptly);
}
}
上面的方法很简单，可以看出来其实就是获取任务，然后执行，如果执行失败则执行相应的处理方法。
接下来我们来看看getTask方法是如何获取任务的：
// 此方法有三种可能：
// 1. 阻塞直到获取到任务返回 我们知道 默认 corePoolSize 之内的线程是不会被回收的 除非主动设置了可以回收
// 2. 超时退出 keepAliveTime 起作用的时候 也就是如果这么多时间内都没有任务 那么应该执行关闭
// 3. 如果发生了以下条件 此方法必须返回 null:
// - 池中有大于 maximumPoolSize 个 workers 存在(通过调用 setMaximumPoolSize 进行设置)
// - 线程池处于 SHUTDOWN 而且 workQueue 是空的 这种不再接受新的任务
// - 线程池处于 STOP 不仅不接受新的线程 连 workQueue 中的线程也不再执行
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 两种可能
// 1. rs == SHUTDOWN && workQueue.isEmpty()
// 2. rs >= STOP
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
// CAS 操作，减少工作线程数
decrementWorkerCount();
return null;
}
boolean timed; // Are workers subject to culling?
for (;;) {
int wc = workerCountOf(c);
// 允许核心线程数内的线程回收 或当前线程数超过了核心线程数 那么有可能发生超时关闭
timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 这里 break，是为了不往下执行后一个 if (compareAndDecrementWorkerCount(c))
// 两个 if 一起看：如果当前线程数 wc > maximumPoolSize 或者超时 都返回 null
// 返回 null 意味着关闭线程。
// 有可能开发者调用了 setMaximumPoolSize() 将线程池的 maximumPoolSize 调小了
// 那么多余的 Worker 就需要被关闭
if (wc <= maximumPoolSize && ! (timedOut && timed))
break;
if (compareAndDecrementWorkerCount(c))
return null;
c = ctl.get(); // Re-read ctl
// compareAndDecrementWorkerCount(c) 失败 线程池中的线程数发生了改变
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
// wc <= maximumPoolSize 同时没有超时
try {
// 到 workQueue 中获取任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
// 如果此 worker 发生了中断 采取的方案是重试
// 如果开发者将 maximumPoolSize 调小了 导致其小于当前的 workers 数量
// 那么意味着超出的部分线程要被关闭 重新进入 for 循环 自然会有部分线程会返回 null
timedOut = false;
}
}
}
到这整个流程基本上就都走了一遍，拒绝策略在线程池中也有四个已经定义好了的实现类可供我们直接使用，如下所示：
// 只要线程池没有被关闭 那么由提交任务的线程自己来执行这个任务
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
// 直接抛出 RejectedExecutionException 异常
// 这个是默认的策略 如果我们构造线程池的时候不传相应的 handler 的话 那就会指定使用这个
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
// 不做任何处理 直接忽略掉这个任务 一般不推荐使用这个
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
// 如果线程池没有被关闭的话 一般不推荐使用这个
// 把队列队头的任务(也就是等待了最长时间的)直接丢弃 然后提交这个任务到等待队列中
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
总结
- 线程池的创建时机（这个是重点，一定要记住）
-
- 如果当前线程数少于 corePoolSize，那么提交任务的时候创建一个新的线程，并由这个线程执行这个任务
- 如果当前线程数已经达到 corePoolSize，那么将提交的任务添加到队列中，等待线程池中的线程去队列中取任务
- 如果队列已满，那么创建新的线程来执行任务，需要保证池中的线程数不会超过 maximumPoolSize，如果此时线程数超过了 maximumPoolSize，那么执行拒绝策略
- 任务执行的过程中发生异常时怎么处理？
-
- 如果某个任务执行出现异常，那么执行任务的线程会被关闭，而不是继续接收其他任务。然后会启动一个新的线程来代替它
- 什么时候执行拒绝策略？
-
- workers 的数量达到了 corePoolSize（任务此时需要进入任务队列），任务入队成功，与此同时线程池被关闭了，而且关闭线程池并没有将这个任务出队，那么执行拒绝策略
- workers 的数量大于等于 corePoolSize，将任务加入到任务队列，可是队列满了，任务入队失败，那么准备开启新的线程，可是线程数已经达到 maximumPoolSize，那么执行拒绝策略