一、简介
ThreadPoolExecutor是Java并发编程中应用最宽泛的类之一，因为线程的创立和销毁须要耗费系统资源，所以通过应用线程池来无效治理线程。
二、工作流程
当向线程池提交一个工作后，线程池是如何来解决的？
如上图：
- 判断外围线程池是否已满。如果不是，则创立一个新的工作线程来执行工作。如果已满，进入下个步骤
- 判断工作队列是否已满。如果工作队列没有满，则将新提交的工作存储在工作队列里。如果队列满了，进入下个步骤
- 判断线程池是否已满。如果没有，则创立一个新的工作线程来执行工作
- 如果线程池已满，则交给饱和策略来解决这个工作
三、源码解读
3.1 线程池参数
ThreadPoolExecutor的结构函数参数含意：
- corePoolSize：外围线程数大小，当线程数<corePoolSize ，会创立线程（即便有闲暇的外围线程也会创立）来执行工作
- maximumPoolSize：最大线程数， 当线程数 >= corePoolSize的时候，会把工作放入工作队列中，当工作队列也满了之后，会再创立新的线程，直到达到最大线程数
- keepAliveTime ：放弃存活工夫，当线程数大于corePoolSize的闲暇线程能放弃存活的最大工夫。
- unit：工夫单位
- workQueue：保留工作的阻塞队列
- threadFactory：创立线程的工厂
- handler：回绝策略 （包含4种：间接抛出异样 – 默认、应用调用者所在线程来运行工作、抛弃队列里最近一个工作来执行当前任务、不解决间接抛弃）
3.2 线程池运行状态、工作线程数量
ThreadPoolExecutor很奇妙的应用了一个AtomicInteger类型的ctl变量，来示意线程池运行状态 和 工作线程数量，高3位用来示意线程池运行状态，低29位用于示意工作线程数量。线程池运行状态有5种，状态和其对应的高3位别离是RUNNING(111)、SHUTDOWN(000)、STOP(001)、TIDYING(010)和TERMINATED(011)。runStateOf()和workerCountOf()办法别离能够计算出线程池运行状态和工作线程数量。源码如下：
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~CAPACITY; }
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }3.3 工作线程Worker
ThreadPoolExecutor外部有一个工作线程Worker类，它并不是继承Thread类，而是外部蕴含了一个Thread对象，而且在初始化该Thread对象时，还将以后Worker对象作为结构函数参数传入，造成了一种互相援用的关系。
工作线程Worker类也是设计的很奇妙，这个类继承了AbstractQueuedSynchronizer且实现了Runnable接口，实现Runnable接口是比拟容易了解的，它的run()办法就是该工作线程的须要做的工作。而为什么要继承AbstractQueuedSynchronizer呢？这是用于后续判断该工作线程是否曾经在执行中，如果正在执行中，那么该工作线程就不能被中断。代码如下
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
// 上面是一些实现AQS须要实现的办法，不一一开展了
}3.4 提交工作
咱们通常应用execute()或submit()来提交工作，那么submit()外部也是应用execute()办法来执行的，所以这里咱们看下execute()办法的源码：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* 分为3个步骤
* 1. 如果以后线程数小于外围线程池大小，会调用addWorker办法，addWorker办法会依据运行状态和线程数量来判断是否须要新增工作线程
* 2. 如果以后线程池运行状态是RUNNING，尝试增加工作到工作队列workQueue中
* 3. 如果无奈增加到工作队列中，持续调用addWorker办法，判断是否新增工作线程
* 4. 最初如果还是无奈新增工作线程，则采取回绝策略
*/
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);
}3.5 新增工作线程
通过下面的代码，很显著咱们接下来就须要看下addWorker()是如何依据运行状态和线程数量来判断是否须要新增工作线程，并且这个办法也是ThreadPoolExecutor中最重要的办法。首先依据运行状态和线程数量来判断是否新增工作线程，如果判断不新增，间接返回false，否则新增工作线程。源码如下：
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
/*
* 1. 如果运行状态是STOP、TIDYING和TERMINATED，那么间接返回false，新增工作线程失败
* 2. 如果运行状态是RUNNING，间接进行下一步
* 3. 如果运行状态是SHUTDOWN，当firstTask不为空（代表新来的工作），也间接返回false，新增工作线程失败。
* firstTask为空的话，如果工作队列为空，也不新增工作线程，反之新增工作线程。
*/
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
/*
* 这里次要是判断工作线程数量是否大于外围线程数（core为true）或大于最大线程数（core为false），如果大于，则增加工作线程失败
* 否则将ctl的值加1，也就是将工作线程数量加1，并且跳出两层循环
*/
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 {
// new一个工作线程
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
// 获取线程池的锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
// 再次查看线程池运行状态
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 {
// 如果新增工作线程失败，将ctl的值减1，也就是将工作线程数量减1
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}3.6 工作线程的工作流程
工作线程Worker的run()办法中的runWroker(this)就是其工作流程，如果firstTask不为空，间接执行firstTask，否则从工作队列中获取工作（如果须要回收工作线程，则应用超时获取工作机制，一旦获取工作超时，则回收工作线程，如果不须要回收工作线程，工作线程将会始终阻塞，直到获取工作），代码如下
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 如果firstTask不为空，间接执行，否则从工作队列中获取工作（超时获取或阻塞获取）
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 {
// 钩子函数，能够继承ThreadPoolExecutor，并重写该办法，通常用来监控工作执行
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 {
// 钩子函数，同beforeExecute()
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
// 工作是否意外终止
completedAbruptly = false;
} finally {
// 退出工作线程
processWorkerExit(w, completedAbruptly);
}
}退出工作线程可分为失常退出或工作异样导致，当completedAbruptly为true时，代表是工作异样导致退出工作线程，completedAbruptly为false，代表工作队列为空，且失常回收线程（当allowCoreThreadTimeOut为false时，线程数量大于外围线程数，或当allowCoreThreadTimeOut为true时，线程数量大于0），代码如下：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 如果工作异样导致工作线程退出，工作线程数减1，如果工作线程失常退出时，在getTask()办法中曾经将工作线程数减1
if (completedAbruptly)
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
// 移除工作线程
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
// 如果运行状态小于STOP，判断是否须要新增工作线程来解决
if (runStateLessThan(c, STOP)) {
// 如果工作线程失常退出
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
// 如果工作线程曾经大于等于min值，那么不须要执行上面的addWork()，间接返回
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// 执行addWorker()来新增工作线程
addWorker(null, false);
}
}3.7 敞开线程池
咱们通常会应用shutdown()或shutdownNow()办法来敞开线程池。
- shutdown是将线程池的状态设置为SHUTDOWN状态，不承受新的工作，正在执行的工作和队列中的工作会继续执行完，如果工作队列为空，则中断并退出闲暇的工作线程。
- shutdownNow将线程池的状态设置成STOP，不承受新的工作，且尝试中断所有正在执行的工作，不执行并返回工作队列中的工作列表。
shutdown()和shutdownNow()办法相似，这里只看shutdown()办法，代码如下：
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 更新运行状态为SHUTDOWN
advanceRunState(SHUTDOWN);
// 中断闲暇的工作线程，因为工作队列为空，所以该工作线程会退出
interruptIdleWorkers();
// 钩子函数
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
/**
* 尝试终止线程池
* 当运行状态为SHUTDOWN且工作队列为空且工作线程数为0，或者运行状态为STOP且工作线程数为0
* 代表线程池敞开，执行terminated函数，这也是个钩子函数
*/
tryTerminate();
}四、总结
以上解读的就是ThreadPoolExecutor中最外围的代码，包含提交工作、新增工作线程、工作线程的工作流程以及线程池的敞开等，了解了这些代码，咱们就能够晓得ThreadPoolExecutor线程池具体是如何工作的，也能在工作中更加得心应手地去应用它。文中若有谬误，欢送斧正。
发表回复