标签：
我之前一篇文章谈到了ThreadPoolExecutor的作用（http://my.oschina.net/xionghui/blog/494004），这篇文章介绍下它的原理，并根据原理分析下它的实现源码。
我们先来查看一下ThreadPoolExecutor API，看看它能实现什么功能，然后看看它是怎么实现这些功能的。
ThreadPoolExecutor API比较长，这里列出几个关键点：
核心和最大池大小：如果运行的线程少于 corePoolSize，则创建新线程来处理请求（即一个Runnable实例），即使其它线程是空闲的。如果运行的线程多于 corePoolSize 而少于 maximumPoolSize，则仅当队列满时才创建新线程。
保持活动时间：如果池中当前有多于 corePoolSize 的线程，则这些多出的线程在空闲时间超过 keepAliveTime 时将会终止。
排队：如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列BlockingQueue，而不添加新的线程。
被拒绝的任务：当 Executor 已经关闭，或者队列已满且线程数量达到maximumPoolSize时（即线程池饱和了），请求将被拒绝。
好，ThreadPoolExecutor实现的功能确实很多，咱们来屡屡ThreadPoolExecutor 的执行过程：
如果运行的线程少于 corePoolSize，ThreadPoolExecutor 会始终首选创建新的线程来处理请求；注意，这时即使有空闲线程也不会重复使用（这和数据库连接池有很大差别）。
如果运行的线程等于或多于 corePoolSize，则 ThreadPoolExecutor 会将请求加入队列BlockingQueue，而不添加新的线程（这和数据库连接池也不一样）。
如果无法将请求加入队列（比如队列已满），则创建新的线程来处理请求；但是如果创建的线程数超出 maximumPoolSize，在这种情况下，请求将被拒绝。
到这儿大家应该了解了线程池的大概执行过程，下面通过源码来介绍下ThreadPoolExecutor是如何实现这些过程和功能的。在理解源码前咱们先来考虑几个问题：
线程池里的线程如何重复利用？比如一个线程执行完请求，怎么控制不退出。
线程池空闲时线程池里的线程数量会不会降到0？
线程池如何保持活动时间？线程可以设置一段时间内闲置就会退出（通过keepAliveTime 设置）。
线程池的阻塞队列有什么用？
请求数量太多如何处理过多的请求？
首先看下线程池的执行过程：
execute(Runnable command)是ThreadPoolExecutor的核心处理方法，用于处理Runnable 请求。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
}
poolSize为线程池内启动的线程数量，当线程池的poolSize小于核心池corePoolSize时，会去执行addIfUnderCorePoolSize(command)，addIfUnderCorePoolSize(Runnable firstTask)会创建一个新线程来处理请求：
private boolean addIfUnderCorePoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < corePoolSize && runState == RUNNING)
t = addThread(firstTask);
} finally {
mainLock.unlock();
}
if (t == null)
return false;
t.start();
return true;
}
可以看到，首先加锁（默认是非公平锁）已保证线程安全，然后会进行double check，状态合法则创建新线程。创建新线程处理任务是通过addThread(Runnable firstTask)方法来完成：
private Thread addThread(Runnable firstTask) {
Worker w = new Worker(firstTask);
Thread t = threadFactory.newThread(w);
if (t != null) {
w.thread = t;
workers.add(w);
int nt = ++poolSize;
if (nt > largestPoolSize)
largestPoolSize = nt;
}
return t;
}
可以看到创建线程时使用了内部类Worker封装了请求Runnable，Worker也是一个Runnable，它封装了firstTask请求，作用后面再介绍。
这里先介绍下通过threadFactory创建新线程的过程：threadFactory是可以自定义的（通过ThreadPoolExecutor 的构造函数传入），默认会使用DefaultThreadFactory，再来看看DefaultThreadFactory是如何创建新线程的：
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
代码很明朗，创建了一个线程，设置为非守护线程并设置优先级为5。其中group和namePrefix是在DefaultThreadFactory的构造函数中定义的：
group = (s != null)? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
现在回到addIfUnderCorePoolSize(Runnable firstTask)方法，创建完线程后会直接start，然后就会调用Work的run()方法，这里介绍下Work的作用：
public void run() {
try {
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task);
task = null;
}
} finally {
workerDone(this);
}
}
其中firstTask就是execute(Runnable command)方法传入的请求，可以看到，如果firstTask不为空，则直接执行，否则会通过getTask()从阻塞队列中获取等待的任务；到这里可以解答第一个问题了：线程池里的线程如何重复利用？一个线程会执行多个请求（即Runnable），当执行完一个请求后会通过getTask()去获取新的请求来执行（是从阻塞队列中获取，后面会介绍）。下面看看getTask()方法：
Runnable getTask() {
for (;;) {
try {
int state = runState;
if (state > SHUTDOWN)
return null;
Runnable r;
if (state == SHUTDOWN) // Help drain queue
r = workQueue.poll();
else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
else
r = workQueue.take();
if (r != null)
return r;
if (workerCanExit()) {
if (runState >= SHUTDOWN) // Wake up others
interruptIdleWorkers();
return null;
}
// Else retry
} catch (InterruptedException ie) {
// On interruption, re-check runState
}
}
}
挑重点介绍：当poolSize小于或等于corePoolSize时，会通过workQueue.take()一直等待，直到workQueue添加新的Runnable，到这里可以解答第二个问题了：线程池空闲时线程池里的线程数量会不会降到0？答案是如果线程池里的线程数量小于或等于核心线程数（corePoolSize）则不会退出任何线程。
然而当poolSize大于corePoolSize时或通过workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)等待keepAliveTime(ns)，这里可以解答第三个问题了：线程池如何保持活动时间?答案是通过阻塞队列workQueue控制。
这里需要注意下，当poolSize大于corePoolSize时且在keepAliveTime内没有获得新的请求，则会判断当前线程是否需要退出，通过workerCanExit()来判断：
private boolean workerCanExit() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
boolean canExit;
try {
canExit = runState >= STOP ||
workQueue.isEmpty() ||
(allowCoreThreadTimeOut &&
poolSize > Math.max(1, corePoolSize));
} finally {
mainLock.unlock();
}
return canExit;
}
从上面可以看到线程退出的条件为：运行状态大于STOP，或者阻塞队列为空，或者当前线程数大于核心线程数。达到条件则返回false，此时getTask()会返回空，然后Work的run()方法里面的while循环则会退出，线程此时会退出并销毁。注意，退出前会执行workerDone(this)进行一些清理操作：
void workerDone(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
if (--poolSize == 0)
tryTerminate();
} finally {
mainLock.unlock();
}
}
介绍完了Work的处理过程咱们再回到execute(Runnable command)方法，前面已经贴出源码了，这里再贴一份：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
}
前面讲到当线程池的poolSize小于核心池corePoolSize时会去创建新线程来执行请求，然后如果poolSize超过了corePoolSize则会直接把请求Runnable添加进阻塞队列workQueue里，这里有两种情况：
1. 如果添加成功，则直接返回。前面介绍过，线程池的线程会执行完自己的请求后会从阻塞队列workQueue中取请求来执行。
2.如果添加失败（比如队列满了），则会通过addIfUnderMaximumPoolSize(command)创建新的线程来处理请求。
到这里可以解答第四个问题了：线程池的阻塞队列有什么用？阻塞队列有两个作用：第一是为了控制线程存活，通过workQueue的take和pull实现；第二是为了存放Runnable对象，以便线程池里空闲的线程处理。
下面继续介绍addIfUnderMaximumPoolSize(command)方法：
private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < maximumPoolSize && runState == RUNNING)
t = addThread(firstTask);
} finally {
mainLock.unlock();
}
if (t == null)
return false;
t.start();
return true;
}
该方法和addIfUnderCorePoolSize(Runnable firstTask)方法类似，大致流程是如果线程池内创建的线程数小于最大线程数maximumPoolSize则创建新线程执行请求，否则返回false。
如果返回false，表示请求数量不能再被处理，此时会调用reject(command)来处理请求：
void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
可以看到，处理过程很简单，就直接调用handler来处理请求；这里的handler可以自定义（同样是通过构造函数传入），handler默认是使用AbortPolicy：
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException();
}
AbortPolicy处理也是很简单粗暴，直接抛出非受查异常RejectedExecutionException。
到这里也可以解答第五个问题了：请求数量太多如何处理过多的请求？答案是通过handler处理的。
ThreadPoolExecutor设置确实十分精巧（作者就是大名鼎鼎的Doug Lea），上面介绍了它的一些实现细节；下面再来谈谈它的一些钩子。
默认情况下，线程池的线程只是在新任务到达时才创建和启动的；如果希望预先启动线程，可以使用方法 prestartCoreThread() 或 prestartAllCoreThreads() 。
prestartCoreThread()会创建并启动一个线程，prestartAllCoreThreads()会启动所以的corePoolSize个线程：
public boolean prestartCoreThread() {
return addIfUnderCorePoolSize(null);
}
public int prestartAllCoreThreads() {
int n = 0;
while (addIfUnderCorePoolSize(null))
++n;
return n;
}
private boolean addIfUnderCorePoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < corePoolSize && runState == RUNNING)
t = addThread(firstTask);
} finally {
mainLock.unlock();
}
if (t == null)
return false;
t.start();
return true;
}
另外还有两个常用的钩子方法：beforeExecute(java.lang.Thread, java.lang.Runnable) 和 afterExecute(java.lang.Runnable, java.lang.Throwable)。
protected void beforeExecute(Thread t, Runnable r) { }
protected void afterExecute(Runnable r, Throwable t) { }
ThreadPoolExecutor内他们的默认实现为空方法。我们可以扩展它们，它们会在执行请求前后调用：
private void runTask(Runnable task) {
final ReentrantLock runLock = this.runLock;
runLock.lock();
try {
if (runState < STOP &&
Thread.interrupted() &&
runState >= STOP)
thread.interrupt();
boolean ran = false;
beforeExecute(thread, task);
try {
task.run();
ran = true;
afterExecute(task, null);
++completedTasks;
} catch (RuntimeException ex) {
if (!ran)
afterExecute(task, ex);
throw ex;
}
} finally {
runLock.unlock();
}
}
我们通常使用 Executors 工厂方法来配置ThreadPoolExecutor，下面摘自ThreadPoolExecutor API：
标签：
原文地址：http://my.oschina.net/xionghui/blog/494698
