标签：
在JUC包中，线程池部分本身有很多组件，可以说是前面所分析的各种技术的一个综合应用。从本文开始，将综合前面的知识，逐个分析线程池的各个组件。
-Executor/Executors
-ThreadPoolExecutor使用介绍
-ThreadPoolExecutor实现原理
–ThreadPoolExecutor的中断与优雅关闭 shutdown + awaitTermination
–shutdown的一个误区
Executor是线程池框架最基本的几个接口：
public interface Executor {
void execute(Runnable command);
}而Executors是线程池框架的一个工具类，利用它可以方便的创建不同策略的线程池：
//单线程线程池：corePoolSize = maxPoolSize = 1, 队列用的LinkedBlockingQueue
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}//固定数目的线程池：corePoolSize = maxPoolSize = n, 队列用的LinkedBlockingQueue
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}//1。CachedThreadPool，corePoolSize = 0, 队列为SynchronousQueue，maxPoolSize = Integer.MAX_VALUE(这也就意味着，每来一个任务，就创建一个线程。
//2。关于SynchronousQueue，后面会单独用一篇来分析。它是个特殊的队列，没本身没有容量，放进去一个，就得等有线程拿出来，才能解除阻塞
//3。从构造参数可以看出，空闲线程，60s没人用，回收
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}//单线程的，具有周期调度功能的线程池
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}
//多线程的，具有周期调度功能的线程池
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
new DelayedWorkQueue());
}从上面可以看出，Executors的各个工具函数，都用的ThreadPoolExecutor/ScheduledThreadPoolExecutor这2个类，下面做详细分析。
下面是ThreadPoolExecutor的参数最全的构造函数，搞清楚了每个参数的含义，也就明白了线程池的各种不同策略，也就明白了上述Executors工具类中的各个工具函数。
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: 线程池始终维护的线程个数
maxPoolSize: corePooSize满了，队列也满的情况下，扩充线程至这个值
keepAliveTime/TimeUnit: maxPoolSize中的空闲线程，过多长时间销毁，总线程数收缩回corePoolSize
blockingQueue: 线程池所用的队列类型
threadFactory: 线程创建工厂，可以自定义，也有一个缺省的
RejectedExecutionHandler: corePoolSize满了，队列满了，maxPoolSize满了，最后的拒绝策略。
从上述构造函数解释，可以看出每次submit的任务，有如下的处理流程：
step1: 判断当前线程数 >= corePoolSize。如果小于，新建线程执行；如果大于，进入step2
step2: 判断队列是否已满。未满，放入；已满，进入step3
step3: 判断当前线程数 >= maxPoolSize。如果小于，新建线程执行；如果大于，进入step4
step4: 根据拒绝策略，拒绝任务
总结一下：先判断corePoolSize, 再判断blockingQueue，再判断maxPoolSize，最后使用拒绝策略
ThreadPoolExecutor的4个内部类，分别定义了4种策略。缺省是AbortPolicy
//策略1：让调用者直接在自己的线程里面执行，线程池不做处理
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
//策略2：线程池直接抛异常
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());
}
}
//策略3：线程池直接把任务丢掉，当作什么也没发生
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
//策略4：把队列里面最老的任务删除掉，把该任务放入队列
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}一般都知道，ThreadPool的基本实现原理就是一个队列 + 一组worker线程，调用中不断往队列中放，worker线程不断去取。但在具体实现中，有不同的实现策略：
策略1： 阻塞队列 vs. 非阻塞队列
在ThreadPoolExecutor中，使用的是阻塞队列，即如下的BlockingQueue接口：
private final BlockingQueue<Runnable> workQueue;这也就意味着，worker内部不需要自己设置wait/notify机制，它只管从队列中取，取的到执行，取不到，自动会阻塞。
也有使用非阻塞队列的，比如Tomcat 6里面的线程池实现（以后会源码详细分析），当没有请求处理时，worker内部自己实现阻塞，然后又新的请求进来，再通知woker。
策略2：新来的请求，是直接放入队列，还是先new一个新的thread？
ThreadPool的处理方式是优先new thread处理，thread count >= corePoolSize的时候，再考虑放入队列。
策略3： 无界队列 vs. 有界队列？
如果无界队列，意味着maxPoolSize的逻辑永远不会执行。这在上面的Executors中，FixedThreadPool已有所体现。
除此之外，还有诸多实现上的细节，下面代码详细分析
//核心结构：一个BlockingQueue + 一个线程的Set + 一把锁（控制对workers, 各种threadCount的互斥访问）
public class ThreadPoolExecutor extends AbstractExecutorService {
。。。
private final BlockingQueue<Runnable> workQueue;
private final ReentrantLock mainLock = new ReentrantLock();
private final HashSet<Worker> workers = new HashSet<Worker>();
。。。
} public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) { //小于corePoolSize的判断
if (runState == RUNNING && workQueue.offer(command)) { //入队列
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command); //进入队列之后，2次检测
}
else if (!addIfUnderMaximumPoolSize(command)) //小于maxPoolSize的判断
reject(command); // 大于maxPoolSize，拒绝请求
}
}
//poolSize < corePoolSize的时候，直接new Thread，加入hashSet
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();
}
return t != null;
}
//队列满了，poolSize < maxPoolSize，再次new thread，加入hashSet
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();
}
return t != null;
}Worker的实现
private final class Worker implements Runnable {
。。。
private Runnable firstTask; //至所以有firstTask这个变量，是因为创建worker的时候，可以直接赋给它一个task执行；有可以不赋给task，让它自己到blockingQueue里面去循环取
Worker(Runnable firstTask) {
this.firstTask = firstTask;
}
//1个死循环，不断从blockingQueue中，取task执行。取不到，就会阻塞在getTask()里面
public void run() {
try {
hasRun = true;
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task);
task = null;
}
} finally {
workerDone(this); //worker线程退出
}
}
。。。
}
//getTask里面有个关键点：当poolSize <= corePoolSize时，是无限期阻塞下去，线程也就会一直存在，不会退出，死掉；当poolSize > corePoolSize或者允许coreThread也死去时，线程就只阻塞keepAliveTime的时间，时间到了，队列还是空的，没有请求，线程就退出，死掉了，同时poolSize--.
Runnable getTask() {
for (;;) {
try {
int state = runState;
if (state > SHUTDOWN)
return null;
Runnable r;
if (state == SHUTDOWN)
r = workQueue.poll(); //poll是非阻塞调用，没有直接返回null
else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS); //等待1个超时时间，默认就是构造函数里面传进去的那个60s
else
r = workQueue.take(); //take是阻塞调用，没有，一直阻塞
if (r != null)
return r;
if (workerCanExit()) {
if (runState >= SHUTDOWN)
interruptIdleWorkers();
return null;
}
} catch (InterruptedException ie) {
// On interruption, re-check runState
}
}
}
volatile int runState;
static final int RUNNING = 0;
static final int SHUTDOWN = 1;
static final int STOP = 2;
static final int TERMINATED = 3;初始处于RUNNING状态，当调用shutdown()之后，切换到SHUTDOWN状态；调用shutdownNow()，切换到STOP状态。
那shutdown与shutdownNow有什么区别吗？
shutdown(): 不会清空队列里面的任务，会等所有任务执行完毕。并且它只会中断那些 > corePoolSize的idle线程
shutdownNow(): 清空队列里面所有任务，同时向所有线程发送中断信号
当队列为空 && pool也为空时，线程池进入Terminated状态。
public void shutdown() {
SecurityManager security = System.getSecurityManager();
if (security != null)
security.checkPermission(shutdownPerm); //权限检查，check当前调用者，是否有权限关闭线程池。没有权限，抛出异常。
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (security != null) {
for (Worker w : workers)
security.checkAccess(w.thread); //权限检查
}
int state = runState;
if (state < SHUTDOWN)
runState = SHUTDOWN; //从running切换到shutdown。不能从stop或者terminated切换到shutdown
try {
for (Worker w : workers) {
w.interruptIfIdle(); //遍历所有线程，向其发送信号
}
} catch (SecurityException se) {
runState = state;
throw se;
}
tryTerminate(); //试图终止线程池
} finally {
mainLock.unlock();
}
}
public List<Runnable> shutdownNow() {
SecurityManager security = System.getSecurityManager();
if (security != null)
security.checkPermission(shutdownPerm);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (security != null) {
for (Worker w : workers)
security.checkAccess(w.thread);
}
int state = runState;
if (state < STOP)
runState = STOP; //切换到stop状态
try {
for (Worker w : workers) {
w.interruptNow(); //变量所有线程，发中断信号，不管是否正在执行任务
}
} catch (SecurityException se) { // Try to back out
runState = state;
// tryTerminate() here would be a no-op
throw se;
}
List<Runnable> tasks = drainQueue(); //清空队列请求
tryTerminate(); // 试图终止线程池
return tasks;
} finally {
mainLock.unlock();
}
}从上面，可以看出，shutdown和shutdownNow的区别有3点：
(1)一个是切换到shutdown状态，一个是切换到stop状态
(2)遍历所有线程，一个调用的interruptIfIdle, 一个调用的interruptNow。
(3)shutdownNow会清空队列中的任务
那interruptIfIdle和interruptNow有什么区别呢？
private final class Worker implements Runnable {
。。。
private final ReentrantLock runLock = new ReentrantLock();
void interruptIfIdle() {
final ReentrantLock runLock = this.runLock;
if (runLock.tryLock()) {
try {
if (hasRun && thread != Thread.currentThread())
thread.interrupt();
} finally {
runLock.unlock();
}
}
}
void interruptNow() {
if (hasRun)
thread.interrupt();
}
public void run() {
try {
hasRun = true;
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) { //getTask内部，也有响应中断的逻辑
runTask(task);
task = null;
}
} finally {
workerDone(this);
}
}
//每次从队列中拿出一个任务，执行之前，会加锁
private void runTask(Runnable task) {
final ReentrantLock runLock = this.runLock;
runLock.lock();
try {
if ((runState >= STOP ||
(Thread.interrupted() && runState >= STOP)) &&
hasRun)
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();
}
}可以看出，interruptIfIdle和interuptNow的关键区别是：前者会加锁访问，这也就意味着，如果被中断的线程，正在执行runTask，则锁是拿不到的。此时shutdown会阻塞，直到woker执行完runTask。
根据上面分析，是不是shutdown一定会阻塞到队列中所有请求都执行完，再返回呢？或者说，shutdown返回的时候，是不是队列里面的请求就一定执行完了呢？
不一定！shutdown返回之后，线程池不一定立即关闭！为什么呢？
请看下面的getTask函数
Runnable getTask() {
for (;;) {
try {
int state = runState;
if (state > SHUTDOWN)
return null;
Runnable r;
if (state == SHUTDOWN) //如果线程池是shutdown状态，就不阻塞了，不管是否能拿到，都是直接返回
r = workQueue.poll(); //关键点：如果是shutdown状态，会一直循环，直到拿空队列里面所有任务
else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
else
r = workQueue.take(); //case1: 别的线程先置了中断标志位，然后当前线程调用take //case 2: 先调用take阻塞在这，然后别的线程置了中断标志位 //2种case，都会抛出异常，进入下面的InterruptedException
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
}
}
}
public void run() {
try {
hasRun = true;
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) { //getTask内部，也有响应中断的逻辑
runTask(task);
task = null;
}
} finally {
workerDone(this);
}
}
//每次从队列中拿出一个任务，执行之前，会加锁
private void runTask(Runnable task) {
final ReentrantLock runLock = this.runLock;
runLock.lock();
try {
if ((runState >= STOP ||
(Thread.interrupted() && runState >= STOP)) &&
hasRun)
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();
}
}总结一下：当执行shutdown的时候，woker线程可能处于以下几种情况：
情况1： 正在执行runTask，此时拿着runLock锁，调用者会阻塞在shutdown上面。
情况2： 正要进入getTask。runTask执行完了，锁释放了，正要去getTask。此时shutdown不会阻塞，状态切换到shutdown状态，就返回了。 而getTask里面，会调用blockingQueue.poll
情况3： 在getTask里面，阻塞在blockQueue.take上面。此时调用shutdown, getTask里面收到中断，再次开始for(;;)循环
情况2，情况3，shutdown不会阻塞，就返回了。
所以不管是shutdown, 还是shutdownNow()，结尾都调用了tryTeminate，下面看看这个函数：
private void tryTerminate() {
if (poolSize == 0) { //线程池里线程没了
int state = runState;
if (state < STOP && !workQueue.isEmpty()) {
state = RUNNING; //关键点：线程池里线程没了，状态是shutdown状态，队列还不为空，此时把状态切会到Running状态。并且重新创建线程，消化队列中的任务
addThread(null);
}
if (state == STOP || state == SHUTDOWN) {
runState = TERMINATED;
termination.signalAll(); //通知awaitTermination函数，不要再等了，线程池关闭
terminated();
}
}
}所以，正确的使用shutdown的方式，应该是如下代码：
executor.shutdown();//只是不能再提交新任务，等待执行的任务不受影响
//调完shutdown，要循环调用awaitTermination，等待线程池真的终止
try {
boolean loop = true;
do { //等待所有任务完成
loop = !executor.awaitTermination(2, TimeUnit.SECONDS); //阻塞，直到线程池里所有任务结束
} while(loop);
} catch (InterruptedException e) {
e.printStackTrace();
} 总结：无论是shutdown，还是shutdownDown，都无法保证线程池立即关闭。他们的本质都只是切换了线程池的状态，发送了中断信号，然后等队列里面的任务为空了，所有线程自己销毁自己。
要让主线程等待线程池彻底终止，需要调用awaitTermination函数。
关于SechduledThreadPoolExecutor，会在接下来的篇章中，详细的单独阐述。
Java多线程 -- JUC包源码分析11 -- ThreadPoolExecutor源码分析
标签：
原文地址：http://blog.csdn.net/chunlongyu/article/details/52487487
