Java 线程池 ThreadPoolExecutor源码简析

如果你对线程池的概念还不甚了解，那么Java 线程池 Executor浅入浅出这篇文章可以让你一步步从0到1了解线程池的搭建过程。关于线程池的更详细的说明，《Java并发编程实战》这本书上讲的很透彻，强烈推荐阅读。

在Executors中为我们提供了多种静态工厂方法来创建各种特性的线程池，其中大多数是返回ThreadPoolExecutor对象。因此本篇博文直接对ThreadPoolExecutor的原理进行剖析，加深对线程池设计的理解。ThreadPoolExecutor有如下两个重要参数：

corePoolSize

线程池核心线程的数量。如果线程池里得线程个数小于corePoolSize，那么就会立即创建一个新线程对象来执行任务。如果任务数大于核心线程数怎么办呢？那么就将任务放在队列中，等待执行。
比如我们去银行办理业务，有若干个人工服务窗口，当来办理业务的客户比较少的时候，可以不用排队，直接去空闲的窗口去办理业务。如果客户比较多了，那么就需要就需要领号排队了。但是服务窗口有限，来办理业务的人太多了，怎么办？maximumPoolSize就派上用场了。
我们需要注意的是，初始化ThreadPoolExecutor的时候，线程Thread不会立即创建和启动，而是等到提交任务的时候才会创建启动，当然调用了prestartAllCoreThreads的话，是可以创建并启动所有Thread.

    /***启动所有核心线程，让他们处于idly状态等待任务的到来。* @return 返回启动的线程的数量*/public int prestartAllCoreThreads() {int n = 0;while (addWorker(null, true))++n;return n;}

另外ThreadPoolExecutor提供了getPoolSize()方法，该方法并不是返回初始化的时候设置的corePoolSize的大小，而是获取线程中实际存在的线程的数量，如果线程池处于TIDYING状态，则返回0：

   //存储正在工作的线程private final HashSet<Worker> workers = new HashSet<>();/*** Returns the current number of threads in the pool.** @return the number of threads*/public int getPoolSize() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {//如果线程处于TIDYING状态，说明线程池里的线程为0return runStateAtLeast(ctl.get(), TIDYING) ? 0: workers.size();} finally {mainLock.unlock();}}

maximumPoolSize

既然有核心线程，那么也有非核心线程，但是这个线程池这个池子不能无限大，maximumPoolSize就限定了线程池中允许的最大线程数。
在线程池阻塞队列（非无界队列）装满了任务之后，如果还有任务加入，如果当前的线程数小于maximumPoolSize，则会新建线程来执行任务。否则就会拒绝执行。
比如去银行排队办理业务的客户越来越多，银行服务大厅已经装不下了。此时如果除了服务窗口之外，银行还有额外的资源可以办理客户的业务，那么就使用额外的资源来处理任务。如果额外的资源也已经满负荷服务客户，对于后面新来的客户，只能“欢迎下次光临”了。

workQueue

用来保存等待执行任务的队列，等待的任务必须实现Runnable接口,线程池与工作队列WorkQueue密切相关，因为并不是一个线程负责处理一个任务，而是一个线程负责处理好多任务。WorkQueue负责保存所有等待执行的任务，而线程池里的线程从工作队列中获取一个任务，然后执行，执行完毕后等待下一个任务。我们先来看看工作线程（下文简称WorkThread)怎么执行任务的，具体的在ThreadPoolExecutorl类的runWorker方法里面，该方法的主要作用就是不断的从WorkQueue中获取任务并执行之：

  final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // 释放锁，允许中断boolean completedAbruptly = true;try {//开启一个循环，从WorkQueue中获取任务并执行之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 interruptif ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {//空方法：任务执行之前的回调beforeExecute(wt, task);Throwable thrown = null;//执行任务task.run();//空方法：任务执行完毕之后的回调afterExecute(task, thrown);}} finally {task = null;//记录当前Worker已经完成任务的数量w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {//使得工作线程退出processWorkerExit(w, completedAbruptly);}}

注意Worker 是一个Runnable，runWorker方法就是在Worker的run方法中执行的。

Worker对象简单说明

Worker是线程池里一个重要的对象，该类还是一个Runnable,该类持有一个Thread，代表着Worker对象在哪个线程中执行，在初始化Worker对象的时候就创建了线程。.

{//由ThreadFactory创建，代表着Worker对象在哪个线程中执行final Thread thread;/** Initial task to run.  Possibly null.Worker对象第一个执行的任务，可能是null，后面在执行任务的时候就要从WorkerQueue里面获取 */Runnable firstTask;/** 记录每个线程完成工作的数量*/volatile long completedTasks;Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;//创建线程this.thread = getThreadFactory().newThread(this);}/** Delegates main run loop to outer runWorker. */public void run() {runWorker(this);}// The value 0 represents the unlocked state.// The value 1 represents the locked state.protected boolean isHeldExclusively() {return getState() != 0;}protected boolean tryAcquire(int unused) {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}protected boolean tryRelease(int unused) {setExclusiveOwnerThread(null);setState(0);return true;}public void lock()        { acquire(1); }public boolean tryLock()  { return tryAcquire(1); }public void unlock()      { release(1); }public boolean isLocked() { return isHeldExclusively(); }void interruptIfStarted() {Thread t;if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {try {t.interrupt();} catch (SecurityException ignore) {}}}}

另外Worker提供了CAS算法进行自旋，提供了lock和unlock等方法，其中1代表着处于locked状态，而0除以非locked状态.关于自旋CAS的基础知识，感兴趣的可以阅读博主的《java线程知识点拾遗(CAS)》。

execute方法

ThreadPoolExecutor提供了execute方法，主要是用新的线程或者线程池里已经有的线程，在未来某个时期执行一个任务，当线程池shutDown或者达到最大容量的时候，任务就不会被执行，此时线程池会执行拒绝策略RejectedExecutionHandler。下面再来大致分析下该方法：

 public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();//如果线程池里的线程小于核心数量if (workerCountOf(c) < corePoolSize) {//创建一个新线程，并将command作为线程执行的第一个任务if (addWorker(command, true))return;c = ctl.get();}//如果线程池处于Running状态，且将任务command成功加入WorkerQueue中if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();//再次校验状态，如果线程池不是处于Running状态，则从队列中移除并执行拒绝策略。if (! isRunning(recheck) && remove(command))reject(command);else if (workerCountOf(recheck) == 0)//注意这是一个很细节的点//线程池没有活动线程了，但是队列里还有任务没执行这种特殊情况，则重新添加一个线程执行。addWorker(null, false);}//队列已经满了，则重新创建线程执行，如果失败则执行拒绝策略else if (!addWorker(command, false))reject(command);}

在这里有一个值得注意的细节，调用workQueue.offer(command)将任务添加到队列之后，如果线程中有正在执行任务的线程，则这个任务会在未来某个时间内执行。但是如果任务入队之后，恰好线程池中没有活跃的线程了，怎么办呢，就再次addWorker，启动新线程执行。（举个不恰当的例子，就好像去即将打烊的饭店去吃饭，你已经扫码付好钱（入队列），然而厨师恰巧都已经准备下班了，不得不重新给你做饭一样），注意此时addWorker的第一个参数是null，因为我们已经将command通过offer添加到队列中了，具体代码段如下：

  else if (workerCountOf(recheck) == 0)//注意这是一个很细节的点//线程池没有活动线程了，但是队列里还有任务没执行这种特殊情况，则重新添加一个线程执行。addWorker(null, false)

可以用如下流程图来大致表示下线程池的工作流程。

addWorker方法简析

addWorker的主要作用就是，在线程池当前的状态和既有的线程边界（corePoolSize,maximumPoolSize)下，新的任务是否能添加成功。

     ** @param firstTask  当Worker被创建的时候（此时线程池里的线程数量小于核心线程），firstTask将作为Worker对象的第一个任务来执行。* 此时firstTask并不会添加到队列中。 或者当队列已经满的情况下，也是直接创建Worker，但是fistTask也不会添加到WorkQueue中去。 */private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {// 外层自旋int c = ctl.get();//获取线程状态int rs = runStateOf(c);// Check if queue empty only if necessary.//如果线程池处于关闭状态，并且WorkerQueue中还有任务在等待执行，则返回falseif (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {//内层自旋int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))//如果线程池内的线程达到最大值，比如达到corePoolSize或者maximunPoolSize，则返回falsereturn false;if (compareAndIncrementWorkerCount(c))// 以CAS的方式尝试把线程数加1// 注意这里只是把线程池中的线程数加1，并没有在线程池中真正的创建线程break retry;c = ctl.get();  // Re-read ctl//如果此时线程池的状态和上次获取的不一样，则继续执行外层自旋if (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}//end inner for}//end outter forboolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {//创建Worker，同时也就创建了Thread对象w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int rs = runStateOf(ctl.get());if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();//将Worker添加到HashSet种   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;}

addWorker工作的大致流程如下：

1、判断线程池状态，如果需要创建Worker对象并同时创建工作线程
2、将此worker添加进workers集合中。
3、启动worker对应的线程，运行worker的run，最终执行了runWorker方法。
4、如果失败了回滚worker的创建动作（addWorkerFailed 方法），即将worker从workers集合中删除，并原子性的减少workerCount。

keepAliveTime的使用原理

该变量表示线程空闲的时间。线程的创建和销毁是需要代价的。线程执行完任务后不会立即销毁，而是继续存活一段时间，存活的时间就是keepAliveTime。默认情况下，该参数只有在线程数大于corePoolSize时才会生效.该变量在从WorkQueue里获取任务的时候会被使用，在前文runWorker里就调用了从workQueue里面获取任务的方法:getTask()：


private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {//开启一个循环int c = ctl.get();int rs = runStateOf(c);//当线程处于SHUTDOWN 状态时，并且workQueue请求队列为空，线程数量的表示-1，并返回一个null。if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}//获取当前工作线程数int wc = workerCountOf(c);//如果allowCoreThreadTimeOut设置为true，或者线程数量大于核心线程数量，keepAliveTime才起作用boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {//减少工作线程if (compareAndDecrementWorkerCount(c))return null;continue;}try {//从阻塞队列中取任务，根据情况选择一直阻塞等待还是超时等待Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();//成功从队列中获取到了任务，并返回之    if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}
}

可以看出，keepAliveTime是在 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)方法中起作用的。在runWorker方法中，开始while循环获取任务，如果超时了就可以跳出while循环，从而Thread完成了其任务而被销毁。：

 final void runWorker(Worker w) {//省略部分代码while (task != null || (task = getTask()) != null) {//省略部分代码}//省略部分代码}