谈谈ThreadPoolExecutor线程池
线程池，凡是学过java的同学都不陌生，一两行简单的代码就能实现并发编程。但java.util.concurrent.ThreadPoolExecutor的源码读起来却是很绕，今天，就让我们来深入了解一下线程池吧。
本文贴出的源码均是基于jdk1.7.0_79版本（jdk1.6版本源代码与该版本代码有比较大差别）。
线程池的构造方法
ThreadPoolExecutor提供了四个构造方法，不过前三个都会转向最后一个构造方法。
源代码1
// 最后一个构造方法
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：池的最大大小，可以理解为临时应急允许达到的最大数量，就像澡堂如果等待的人实在太多，就再开放一些淋浴位置。不过一般这个字段没用，我们使用线程池就是想有一个确定的最大线程运行数量，这个值一般和核心池大小值一致。
keepAliveTime：当线程池中的线程没有可以运行的新任务时，等待多久后自行结束，就像澡堂的淋浴喷头在等待几秒钟后如果没有人洗澡就会自动关闭。
workQueue：等待队列，当线程池运行的线程数量达到了核心池大小，新来的任务就会进到这里，就像澡堂里会有一个休息室。Executors.newFixedThreadPool(int)方法默认使用LinkedBlockingQueue<Runable>。
threadFactory：构造Thread的工厂，要实现的方法是Thread newThread(Runnable r)，就像澡堂中创建一个淋浴位置一样，可以给淋浴位置设置一下名称，组别之类的。
handler：当线程池线程数满了，等待队列也满了，也达到了临时应急设置的最大线程数量的时候，要执行的方法，就像澡堂洗澡的人满了，等待的人满了，澡堂该如何处理新进来要洗澡的人。
RejectedExecutionHandler接口的方法是rejectedExecution(Runnabler, ThreadPoolExecutor e)，java默认实现了四种拒绝执行处理器，并默认使用AbortPolicy处理器。
- CallerRunsPolicy：如果线程池没有关闭，就执行这个任务。也就是说会在运行线程池提交任务方法的线程中运行线程。
- AbortPolicy：抛出一个
RejectedExecutionException异常。 - DiscardPolicy：忽略这个任务，并且什么也不做。
- DiscardOldestPolicy：立刻取出等待队列中头一个任务（也就是等待时间最长的任务），然后再由线程池执行新任务。
向线程池提交任务
在第一节中，我提到了线程和任务，一定要区分这两个概念。使用者提交的是任务，可以在线程池中运行，或者被放到等待队列中。线程是线程池创建的，使用者的任务是被线程执行的。
我们来看看提交任务的源代码吧
源代码2
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}翻译一下源代码中的注释：
- 如果线程池的线程数少于corePoolSize大小，尝试创建一个线程并执行提交的任务。addWorker方法会检查线程池的状态是不是可以提交任务，并且检查线程数量。通常它不会返回false。
- 如果任务可以加入到等待队列中，那么仍需检查我们是否应该添加一个线程，因为自提交任务到添加到队列中间有可能有线程已经退出。如果有必要，就从队列中删除该任务，并拒绝该任务的执行。或者如果有线程退出，则开始一个空任务新线程。
- 如果我们不能把任务加入到等待队列，则再次尝试开始一个带有该任务的新线程。如果失败，则说明等待队列满了，线程池数量等于允许的最大线程池数量，因此拒绝该任务的执行。
源代码2看似简单，其实还有些难以理解。我们根据注释的三部分来分别理解一下。
第一部分
先判断线程池的线程数量是不是小于设定的corePoolSize。当条件成立时，调用addWorker方法，返回结果为true就结束。
第二部分
说明线程池的线程数量大于等于corePoolSize，或者调用addWorker方法，返回结果为false。
这时，会先判断线程池是否还在运行，并且尝试把任务添加到等待队列中。
如果能加到等待队列中，会再次判断线程池是否还在运行。如果不在运行了，就移除添加到等待队列中的任务。并且调用rejectHandler拒绝该任务执行。
如果线程池还在运行，或者移除等待队列中的任务失败，则再判断线程池的线程数量是不是等于0，如果等于0，就创建一个空线程。
为什么要进行上边这个步骤？而不是直接添加到等待队列中就结束？原因是workQueue的offer方法执行是有时间的。
- 当offer成功后，可能已经过去了几十毫秒，这时线程池如果停止运行，应当把等待队列中的任务清除调并执行拒绝方法。
- 还有一种可能，offer过去了几十毫秒，线程池中的线程都已经运行完成结束了，没有线程能够从等待队列中取任务了，这个时候应该创建一个空线程。
第三部分
说明线程池的线程数量大于等于corePoolSize。或者调用addWorker方法，返回结果为false后，线程池停止运行了或者等待队列满了。
这是什么意思呢？线程池的线程数量大于等于corePoolSize了，等待队列满了，或者说线程池停止运行了。这时候能干啥呢？尝试创建一个带任务新线程，如果还失败，说明达到最大允许的最大线程池数量，只能拒绝方法运行了。
线程池创建线程执行任务
在源代码2中，出现了3次的addWorker方法是创建线程的关键。我们来看一下addWorker方法的源码。
源代码3
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
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;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
final ReentrantLock mainLock = this.mainLock;
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int c = ctl.get();
int rs = runStateOf(c);
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
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;
}源代码3有点长，我们先来看上半部分for循环。
//Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;注释中说了，在必要时检查等待队列是否为空。意思是，在运行状态为shutdown，firstTask为null，等待队列为空的时候，返回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;
// else CAS failed due to workerCount change; retry inner loop
}
这段代码中，先会判断正在运行的线程数量有没有超过指定corePoolSize，注意，入参的第二个参数在这里用到，当core=true时，是和corePoolSize判断，core=false时，是和maximumPoolSize判断。
判断如果没超过，尝试使用cas更新线程池数量值。更新成功就开始真正的新增线程，否则就继续循环。
源代码3的下半部分比较简单，会先创建一个Worker，接着对代码段加锁。因为加锁会耗时，所以加完锁后再判断是否线程池停止运行。没有的话，将Worker添加到HashSet类型的workers中，解锁。解锁完成后，启动这个Worker，结束。
再看Worker
通读上边两节，我们成功地区分了线程和任务。也知道了线程被包裹在Worker中放到了HashSet中。那Worker到底是什么？Worker运行完后怎么从HashSet类型的workers中退出呢？来看源码吧！
源代码4
// 构造函数
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
// 实现了Runnable接口
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}Worker是线程池中的内部类，实现了Runnable接口。根据源代码4可以看出，在源代码3中拿到的Worker.thread中包装的任务就是Worker自己，在源代码3中运行的也就是Worker的run方法。
源代码4中提到了该run方法实际上代理了实际任务。runWorker方法是运行实际任务的入口，来继续看源码吧！
源代码5
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}略读源代码5，我们发现这里有一个while循环，尝试从Worker本身的firstTask和getTask()方法中获取可执行的任务。什么意思呢？也就是说当你在源代码2中调用addWorker时，第一个参数就是firstTask，运行Worker时，也就会先运行firstTask，运行完firstTask后才会调用getTask()方法从别的地方获取新任务。
其他地方需要注意的是，java.util.concurrent.ThreadPoolExecutor在这里给我们暴露出了两个方法，方便我们观察任务执行。他们分别是：
protected void beforeExecute(Thread t, Runnable r) { }
protected void afterExecute(Runnable r, Throwable t) { }你可以自己继承线程池，重写这两个方法。线程池的作者给了一个例子，允许外部暂停线程池执行任务。
在方法processWorkerExit中，线程池的workers会移除当前Worker。
getTask()方法是我们最后要追寻的方法，下面是源码。
源代码6
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
boolean timed; // Are workers subject to culling?
for (;;) {
int wc = workerCountOf(c);
timed = allowCoreThreadTimeOut || wc > corePoolSize;
if (wc <= maximumPoolSize && ! (timedOut && timed))
break;
if (compareAndDecrementWorkerCount(c))
return null;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}其中我们看到，实际上会从线程池的等待队列workQueue中获取任务，如果线程池的线程数量超过corePoolSize，那么从等待队列中等待指定超时时间获取任务（poll），否则被等待队列调起（take）。这两个方法分别会让线程陷入timed_waiting状态和waiting状态。
为什么这么设计？原因是当线程池线程数量超过corePoolSize后，需要在尽量短的时间内销毁掉多余的线程，如果从等待队列中等待指定时间没有获取到任务，说明资源不紧张了，可以销毁线程了。
流程图
看了这么多是不是还有点头晕，这里我简单画了一个提交任务的流程图和runWorker运行任务的流程图，供大家参考理解。
