前言:
线程是稀缺资源，如果被无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，合理的使用线程池对线程进行统一分配、调优和监控，有以下好处：
1、降低资源消耗；
2、提高响应速度；
3、提高线程的可管理性。
线程池用一个32位的int来同时保存runState和workerCount，其中高3位是runState，其余29位是workerCount。代码中会反复使用runStateOf和workerCountOf来获取runState和workerCount。
2184951-16c59af5d9d66527.png
1、Executors.newFixedThreadPool(10)初始化一个包含10个线程的线程池executor；
2、通过executor.execute方法提交20个任务，每个任务打印当前的线程名；
3、负责执行任务的线程的生命周期都由Executor框架进行管理；
ThreadPoolExecutor
Executors是java线程池的工厂类，通过它可以快速初始化一个符合业务需求的线程池，如Executors.newFixedThreadPool方法可以生成一个拥有固定线程数的线程池。
2184951-35585a83d35ee516.png
其本质是通过不同的参数初始化一个ThreadPoolExecutor对象，具体参数描述如下：
corePoolSize：
线程池中的核心线程数，当提交一个任务时，线程池创建一个新线程执行任务，直到当前线程数等于corePoolSize；如果当前线程数为corePoolSize，继续提交的任务被保存到阻塞队列中，等待被执行；如果执行了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有核心线程。
maximumPoolSize：
线程池中允许的最大线程数。如果当前阻塞队列满了，且继续提交任务，则创建新的线程执行任务，前提是当前线程数小于maximumPoolSize；
keepAliveTime：
线程空闲时的存活时间，即当线程没有任务执行时，继续存活的时间；默认情况下，该参数只在线程数大于corePoolSize时才有用；
unit：
keepAliveTime的单位；
workQueue：
用来保存等待被执行的任务的阻塞队列，且任务必须实现Runable接口，在JDK中提供了如下阻塞队列：
1、ArrayBlockingQueue：基于数组结构的有界阻塞队列，按FIFO排序任务；
2、LinkedBlockingQuene：基于链表结构的阻塞队列，按FIFO排序任务，吞吐量通常要高于ArrayBlockingQuene；
3、SynchronousQuene：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQuene；
4、priorityBlockingQuene：具有优先级的无界阻塞队列；
threadFactory：
创建线程的工厂，通过自定义的线程工厂可以给每个新建的线程设置一个具有识别度的线程名。
2184951-d2d8fd007c7f7a27.png
handler：
线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，线程池提供了4种策略：
1、AbortPolicy：直接抛出异常，默认策略；
2、CallerRunsPolicy：用调用者所在的线程来执行任务；
3、DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
4、DiscardPolicy：直接丢弃任务；
当然也可以根据应用场景实现RejectedExecutionHandler接口，自定义饱和策略，如记录日志或持久化存储不能处理的任务。
Exectors：
Exectors工厂类提供了线程池的初始化接口，主要有如下几种：
newFixedThreadPool
2184951-35585a83d35ee516.png
初始化一个指定线程数的线程池，其中corePoolSize == maximumPoolSize，使用LinkedBlockingQuene作为阻塞队列，不过当线程池没有可执行任务时，也不会释放线程。
newCachedThreadPool
2184951-9b76630ac48f318c.png
1、初始化一个可以缓存线程的线程池，默认缓存60s，线程池的线程数可达到Integer.MAX_VALUE，即2147483647，内部使用SynchronousQueue作为阻塞队列；
2、和newFixedThreadPool创建的线程池不同，newCachedThreadPool在没有任务执行时，当线程的空闲时间超过keepAliveTime，会自动释放线程资源，当提交新任务时，如果没有空闲线程，则创建新线程执行任务，会导致一定的系统开销；
所以，使用该线程池时，一定要注意控制并发的任务数，否则创建大量的线程可能导致严重的性能问题
newSingleThreadExecutor
2184951-deded05302aaf255.png
初始化的线程池中只有一个线程，如果该线程异常结束，会重新创建一个新的线程继续执行任务，唯一的线程可以保证所提交任务的顺序执行，内部使用LinkedBlockingQueue作为阻塞队列。
newScheduledThreadPool
2184951-14823f17e4e4a09e.png
初始化的线程池可以在指定的时间内周期性的执行所提交的任务，在实际的业务场景中可以使用该线程池定期的同步数据。
实现原理
除了newScheduledThreadPool的内部实现特殊一点之外，其它几个线程池都是基于ThreadPoolExecutor类实现的。
线程池内部状态
2184951-5a620e0f56cbb008.png
其中AtomicInteger变量ctl的功能非常强大：利用低29位表示线程池中线程数，通过高3位表示线程池的运行状态：
1、RUNNING：-1 << COUNT_BITS，即高3位为111，该状态的线程池会接收新任务，并处理阻塞队列中的任务；
2、SHUTDOWN： 0 << COUNT_BITS，即高3位为000，该状态的线程池不会接收新任务，但会处理阻塞队列中的任务；
3、STOP ： 1 << COUNT_BITS，即高3位为001，该状态的线程不会接收新任务，也不会处理阻塞队列中的任务，而且会中断正在运行的任务；
4、TIDYING ： 2 << COUNT_BITS，即高3位为010；
5、TERMINATED： 3 << COUNT_BITS，即高3位为011；
线程池状态默认从RUNNING开始流转，到状态TERMINATED结束，中间不需要经过每一种状态，但不能让状态回退。下面是状态变化可能的路径和变化条件：
3294095-6a507005d933355b.png
任务提交
线程池框架提供了两种方式提交任务，根据不同的业务需求选择不同的方式。
Executor.execute()
2184951-834971c24d085d31.png
通过Executor.execute()方法提交的任务，必须实现Runnable接口，该方式提交的任务不能获取返回值，因此无法判断任务是否执行成功。
ExecutorService.submit()
2184951-ea9fe289ca3de89f.png
通过ExecutorService.submit()方法提交的任务，可以获取任务执行完的返回值。
任务执行
当向线程池中提交一个任务，线程池会如何处理该任务？
execute实现：
调用execute将会根据线程池的情况创建Worker，可以归纳出下图四种情况：
3294095-c3b458db36a11d5e.png
标记1对应第一种情况，要留意addWorker传入了core，core=true为corePoolSize，core=false为maximumPoolSize，新增时需要检查workerCount是否超过允许的最大值。
标记2对应第二种情况，检查线程池是否在运行，并且将任务加入等待队列。标记3再检查一次线程池状态，如果线程池忽然处于非运行状态，那就将等待队列刚加的任务删掉，再交给RejectedExecutionHandler处理。标记4发现没有worker，就先补充一个空任务的worker。
标记5对应第三种情况，等待队列不能再添加任务了，调用addWorker添加一个去处理。
标记6对应第四种情况，addWorker的core传入false，返回调用失败，代表workerCount已经超出maximumPoolSize，那就交给RejectedExecutionHandler处理。
addWorker实现
从方法execute的实现可以看出：addWorker主要负责创建新的线程并执行任务，代码实现如下：
private boolean addWorker(Runnable firstTask, boolean core) {
//1
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
}
}
//2
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
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 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;
}
在addWoker的标记1部分：
1、判断线程池的状态，如果线程池的状态值大于或等SHUTDOWN，则不处理提交的任务，直接返回；
2、通过参数core判断当前需要创建的线程是否为核心线程，如果core为true，且当前线程数小于corePoolSize，则跳出循环，开始创建新的线程，具体实现在标记2部分。
在标记2部分中，线程池的工作线程通过Woker类实现，在ReentrantLock锁的保证下，把Woker实例插入到HashSet后，并启动Woker中的线程，其中Worker类设计如下：
1、继承了AQS类，可以方便的实现工作线程的中止操作；
2、实现了Runnable接口，可以将自身作为一个任务在工作线程中执行；
3、当前提交的任务firstTask作为参数传入Worker的构造方法；
Worker的执行
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
从Woker类的构造方法实现可以发现：线程工厂在创建线程thread时，将Woker实例本身this作为参数传入，当执行start方法启动线程thread时，本质是执行了Worker的runWorker方法。
runWorker实现：
2184951-1e8ed00138c189ea.png
runWorker方法是线程池的核心：
1、线程启动之后，通过unlock方法释放锁，设置AQS的state为0，表示运行中断；
2、获取第一个任务firstTask，执行任务的run方法，不过在执行任务之前，会进行加锁操作，任务执行完会释放锁；
3、在执行任务的前后，可以根据业务场景自定义beforeExecute和afterExecute方法；
4、firstTask执行完成之后，通过getTask方法从阻塞队列中获取等待的任务，如果队列中没有任务，getTask方法会被阻塞并挂起，不会占用cpu资源；
getTask实现：
2184951-a63a6646c456f715.png
整个getTask操作在自旋下完成：
1、workQueue.take：如果阻塞队列为空，当前线程会被挂起等待；当队列中有任务加入时，线程被唤醒，take方法返回任务，并执行；
2、workQueue.poll：如果在keepAliveTime时间内，阻塞队列还是没有任务，则返回null；
所以，线程池中实现的线程可以一直执行由用户提交的任务。
Future和Callable实现
通过ExecutorService.submit()方法提交的任务，可以获取任务执行完的返回值。
2184951-cdacf86769a61288.png
在实际业务场景中，Future和Callable基本是成对出现的，Callable负责产生结果，Future负责获取结果。
1、Callable接口类似于Runnable，只是Runnable没有返回值。
2、Callable任务除了返回正常结果之外，如果发生异常，该异常也会被返回，即Future可以拿到异步执行任务各种结果；
3、Future.get方法会导致主线程阻塞，直到Callable任务执行完成；
submit实现：
2184951-15281a586352e509.png
通过submit方法提交的Callable任务会被封装成了一个FutureTask对象。
FutureTask：
2184951-d109631c519443cc.png
1、FutureTask在不同阶段拥有不同的状态state，初始化为NEW；
2、FutureTask类实现了Runnable接口，这样就可以通过Executor.execute方法提交FutureTask到线程池中等待被执行，最终执行的是FutureTask的run方法；
FutureTask.get实现:
2184951-62fdfdd3f3a14e22.png
内部通过awaitDone方法对主线程进行阻塞，具体实现如下：
2184951-aa841730137097ac.png
1、如果主线程被中断，则抛出中断异常；
2、判断FutureTask当前的state，如果大于COMPLETING，说明任务已经执行完成，则直接返回；
3、如果当前state等于COMPLETING，说明任务已经执行完，这时主线程只需通过yield方法让出cpu资源，等待state变成NORMAL；
4、通过WaitNode类封装当前线程，并通过UNSAFE添加到waiters链表；
5、最终通过LockSupport的park或parkNanos挂起线程；
FutureTask.run实现:
2184951-29642db6d9c6589e.png
FutureTask.run方法是在线程池中被执行的，而非主线程
1、通过执行Callable任务的call方法；
2、如果call执行成功，则通过set方法保存结果；
3、如果call执行有异常，则通过setException保存异常；
set:
2184951-19a5160726da146d.png
setException:
2184951-34a8bba028ba58be.png
set和setException方法中，都会通过UnSAFE修改FutureTask的状态，并执行finishCompletion方法通知主线程任务已经执行完成；
finishCompletion:
2184951-c380e58cdcb02e31.png
1、执行FutureTask类的get方法时，会把主线程封装成WaitNode节点并保存在waiters链表中；
2、FutureTask任务执行完成后，通过UNSAFE设置waiters的值，并通过LockSupport类unpark方法唤醒主线程；
最后来看结束worker需要执行的操作：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
//1
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
//2
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
//3
tryTerminate();
int c = ctl.get();
//4
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
正常情况下，在getTask里就会将workerCount减一。标记1处用变量completedAbruptly判断worker是否异常退出，如果是，需要补充对workerCount的减一。
标记2将worker处理任务的数量累加到总数，并且在集合workers中去除。
标记3尝试终止线程池，后续会研究。
标记4处理线程池还是RUNNING或SHUTDOWN状态时，如果worker是异常结束，那么会直接addWorker。如果allowCoreThreadTimeOut=true，并且等待队列有任务，至少保留一个worker；如果allowCoreThreadTimeOut=false，workerCount不少于corePoolSize。
线程池的关闭
线程池的关闭不是一关了事，worker在池里处于不同状态，必须安排好worker的"后事"，才能真正释放线程池。ThreadPoolExecutor提供两种方法关闭线程池：
shutdown：不能再提交任务，已经提交的任务可继续运行；
shutdownNow：不能再提交任务，已经提交但未执行的任务不能运行，在运行的任务可继续运行，但会被中断，返回已经提交但未执行的任务。
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess(); //1 安全策略机制
advanceRunState(SHUTDOWN); //2
interruptIdleWorkers(); //3
onShutdown(); //4 空方法，子类实现
} finally {
mainLock.unlock();
}
tryTerminate(); //5
}
shutdown将线程池切换到SHUTDOWN状态，并调用interruptIdleWorkers请求中断所有空闲的worker，最后调用tryTerminate尝试结束线程池。
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
interruptWorkers();
tasks = drainQueue(); //1
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
shutdownNow和shutdown类似，将线程池切换为STOP状态，中断目标是所有worker。drainQueue会将等待队列里未执行的任务返回。
interruptIdleWorkers和interruptWorkers实现原理都是遍历workers集合，中断条件符合的worker。
上面的代码多次出现调用tryTerminate，这是一个尝试将线程池切换到TERMINATED状态的方法。
final void tryTerminate() {
for (;;) {
int c = ctl.get();
//1
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
//2
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
//3
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
标记1检查线程池状态，下面几种情况，后续操作都没有必要，直接return。
RUNNING（还在运行，不能停）
TIDYING或TERMINATED（已经没有在运行的worker）
SHUTDOWN并且等待队列非空（执行完才能停）
标记2在worker非空的情况下又调用了interruptIdleWorkers，你可能疑惑在shutdown时已经调用过了，为什么又调用，而且每次只中断一个空闲worker？你需要知道，shutdown时worker可能在执行中，执行完阻塞在队列的take，不知道要结束，所有要补充调用interruptIdleWorkers。每次只中断一个是因为processWorkerExit时，还会执行tryTerminate，自动中断下一个空闲的worker。
标记3是最终的状态切换。线程池会先进入TIDYING状态，再进入TERMINATED状态，中间提供了terminated这个空方法供子类实现。
调用关闭线程池方法后，需要等待线程池切换到TERMINATED状态。awaitTermination检查限定时间内线程池是否进入TERMINATED状态，代码如下：
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (;;) {
if (runStateAtLeast(ctl.get(), TERMINATED))
return true;
if (nanos <= 0)
return false;
nanos = termination.awaitNanos(nanos);
}
} finally {
mainLock.unlock();
}
}
参考文章：
http://www.jianshu.com/p/87bff5cc8d8c
http://www.jianshu.com/p/f62a3f452869
网友评论