点赞再看，养成习惯，搜一搜【一角钱技术】关注更多原创技术文章。本文 GitHub org_hejianhui/JavaStudy 已收录，有我的系列文章。
前言
在介绍线程池之前，我们先回顾下线程的基本知识。其中线程池包括ThreadPoolExecutor 默认线程和ScheduledThreadPoolExecutor 定时线程池 ，本篇重点介绍ThreadPoolExecutor线程池。
线程
线程是调度CPU资源的最小单位，线程模型分为KLT模型与ULT模型，JVM使用的是KLT模型，Java线程与OS线程保持 1:1 的映射关系，也就是说有一个Java线程也会在操作系统里有一个对应的线程。
内核线程模型
内核线程(KLT)：系统内核管理线程(KLT)，内核保存线程的状态和上下文信息，线程阻塞不会引起进程阻塞。在多处理器系统上，多线程在多处理器上并行运行。线程的创建、调度和管理由内核完成，效率比ULT要慢，比进程操作快。
用户线程模型
用户线程(ULT)：用户程序实现，不依赖操作系统核心，应用提供创建、同步、调度和管理线程的函数来控制用户线程。不需要用户态/内核态切换，速度快。内核对ULT无感知，线程阻塞则进程（包括它的所有线程）阻塞。
Java线程生命状态
Java线程有多种生命状态：
- NEW ，新建
- RUNNABLE ，运行
- BLOCKED ，阻塞
- WAITING ，等待
- TIMED_WAITING ，超时等待
- TERMINATED，终结
状态切换如下图所示：
Java线程实现方式
Java线程实现方式主要有四种：
- 继承Thread类
- 实现Runnable接口、
- 实现Callable接口通过FutureTask包装器来创建Thread线程、
- 使用ExecutorService、Callable、Future实现有返回结果的多线程。
其中前两种方式线程执行完后都没有返回值，后两种是带返回值的。
继承Thread类创建线程
Thread类本质上是实现了Runnable接口的一个实例，代表一个线程的实例。启动线程的唯一方法就是通过Thread类的start()实例方法。start()方法是一个native方法，它将启动一个新线程，并执行run()方法。这种方式实现多线程很简单，通过自己的类直接extend Thread，并复写run()方法，就可以启动新线程并执行自己定义的run()方法。例如：
public class MyThread extends Thread {
public void run() {
System.out.println("关注一角钱技术，获取Java架构资料");
}
}
MyThread myThread1 = new MyThread();
MyThread myThread2 = new MyThread();
myThread1.start();
myThread2.start();
实现Runnable接口创建线程
如果自己的类已经extends另一个类，就无法直接extends Thread，此时，可以实现一个Runnable接口，如下：
// 实现Runnable接口的类将被Thread执行，表示一个基本的任务
public interface Runnable {
// run方法就是它所有的内容，就是实际执行的任务
public abstract void run();
}
public class MyThread implements Runnable {
public void run() {
System.out.println("关注一角钱技术，获取Java架构资料");
}
}
为了启动MyThread，需要首先实例化一个Thread，并传入自己的MyThread实例：
MyThread myThread = new MyThread();
Thread thread = new Thread(myThread);
thread.start();
事实上，当传入一个Runnable target参数给Thread后，Thread的run()方法就会调用target.run()，参考JDK源代码：
public void run() {
if (target != null) {
target.run();
}
}
实现Callable接口通过FutureTask包装器来创建Thread线程
Callable接口（也只有一个方法）定义如下：
public interface Callable<V> {
V call（） throws Exception;
}
//Callable同样是任务，与Runnable接口的区别在于它接收泛型，同时它执行任务后带有返回内容
public class SomeCallable<V> implements Callable<V> {
// 相对于run方法的带有返回值的call方法
@Override
public V call() throws Exception {
// TODO Auto-generated method stub
return null;
}
}
Callable<V> oneCallable = new SomeCallable<V>();
//由Callable<Integer>创建一个FutureTask<Integer>对象：
FutureTask<V> oneTask = new FutureTask<V>(oneCallable);
//注释：FutureTask<Integer>是一个包装器，它通过接受Callable<Integer>来创建，它同时实现了Future和Runnable接口。
//由FutureTask<Integer>创建一个Thread对象：
Thread oneThread = new Thread(oneTask);
oneThread.start();
//至此，一个线程就创建完成了。
使用ExecutorService、Callable、Future实现有返回结果的线程
ExecutorService、Callable、Future三个接口实际上都是属于Executor框架。返回结果的线程是在JDK1.5中引入的新特征，有了这种特征就不需要再为了得到返回值而大费周折了。而且自己实现了也可能漏洞百出。（下部分来讲线程池了）
- 可返回值的任务必须实现Callable接口。
- 类似的，无返回值的任务必须实现Runnable接口。
执行Callable任务后，可以获取一个Future的对象，在该对象上调用get就可以获取到Callable任务返回的Object了。
注意：get方法是阻塞的，即：线程无返回结果，get方法会一直等待。
再结合线程池接口ExecutorService就可以实现传说中有返回结果的多线程了。
下面提供了一个完整的有返回结果的多线程测试例子。代码如下：
package com.niuh.thread.v4;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
* <p>
* 使用ExecutorService、Callable、Future实现有返回结果的线程
* </p>
*/
public class MyThread {
public static void main(String[] args) throws ExecutionException,
InterruptedException {
System.out.println(("----程序开始运行----"));
Date date1 = new Date();
int taskSize = 5;
// 创建一个线程池
ExecutorService pool = Executors.newFixedThreadPool(taskSize);
// 创建多个有返回值的任务
List<Future> list = new ArrayList<Future>();
for (int i = 0; i < taskSize; i++) {
Callable c = new MyCallable(i + " ");
// 执行任务并获取Future对象
Future f = pool.submit(c);
// System.out.println(">>>" + f.get().toString());
list.add(f);
}
// 关闭线程池
pool.shutdown();
// 获取所有并发任务的运行结果
for (Future f : list) {
// 从Future对象上获取任务的返回值，并输出到控制台
System.out.println(">>>" + f.get().toString());
}
Date date2 = new Date();
System.out.println("----程序结束运行----，程序运行时间【"
+ (date2.getTime() - date1.getTime()) + "毫秒】");
}
}
class MyCallable implements Callable<Object> {
private String taskNum;
MyCallable(String taskNum) {
this.taskNum = taskNum;
}
public Object call() throws Exception {
System.out.println(">>>" + taskNum + "任务启动");
Date dateTmp1 = new Date();
Thread.sleep(1000);
Date dateTmp2 = new Date();
long time = dateTmp2.getTime() - dateTmp1.getTime();
System.out.println(">>>" + taskNum + "任务终止");
return taskNum + "任务返回运行结果,当前任务时间【" + time + "毫秒】";
}
}
协程
协程（纤程，用户级线程），目的是为了追求最大力度的发挥硬件性能和提升软件的速度，协程基本原理是：在某个点挂起当前的任务，并且保存栈信息，去执行另一个任务；等完成或达到某个条件时，再还原原来的栈信息并继续执行（整个过程不需要上下文切换）。
协程的概念很早就提出来了，但直到最近几年才在某些语言（如Lua）中得到广泛应用。
协程的目的：当我们在使用多线程的时候，如果存在长时间的I/O操作。这个时候线程一直处于阻塞状态，如果线程很多的时候，会存在很多线程处于空闲状态，造成了资源应用不彻底。相对的协程不一样了，在单线程中多个任务来回执行如果出现长时间的I/O操作，让其让出目前的协程调度，执行下一个任务。当然可能所有任务，全部卡在同一个点上，但是这只是针对于单线程而言，当所有数据正常返回时，会同时处理当前的I/O操作。
Java原生不支持协程，在纯java代码里需要使用协程的话需要引入第三方包,如：quasar
<dependency>
<groupId>co.paralleluniverse</groupId>
<artifactId>quasar-core</artifactId>
<version>0.8.0</version>
<classifier>jdk8</classifier>
</dependency>
线程池
“线程池”，顾名思义就是一个线程缓存，线程是稀缺资源，如果被无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，因此 Java 中提供线程池对线程进行统一分配、调优和监控。
线程池介绍
在web开发中，服务器需要接受并处理请求，所以会为一个请求分配一个线程来进行处理。如果每次请求都创建一个线程的话实现起来非常简单，但是存在一个问题：如果并发的请求数量非常多，但每个线程执行的时间很短，这样就会频繁的创建和销毁线程，如此一来会大大降低系统的效率。可能出现服务器在为每个请求创建新线程和销毁线程上花费的时间和消耗的系统资源要比处理实际的用户请求的时间和资源更多。
那么有没有一种办法使执行完一个任务，并不被销毁，而是可以继续执行其他的任务呢？
这就是线程池的目的。线程池为线程生命周期的开销和资源不足问题提供了解决方案。通过对多个任务重用线程，线程创建的开销被分摊到多个任务上。
什么时候使用线程池？
- 单个任务处理时间比较短；
- 需要处理的任务数量很大。
线程池优势
- 重用存在的线程。减少线程黄金、消亡的开销，提高性能；
- 提高响应速度。当任务到达时，任务可以不需要等待线程创建就能立即执行；
- 提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配、调优和监控。
Executor框架
Executor接口是线程池框架中最基础的部分，定义来一个用于执行 Runnable 的 execute 方法。下面为它的继承与实现
ExecutorService接口
从图中可以看出 Executor 下有一个重要的子接口 ExecutorService ，其中定义来线程池的具体行为
- execute(Runnable command)：履行Ruannable类型的任务；
- submit(task)：可用来提交Callable或Runnable任务，并返回代表此任务的Future对象；
- shutdown()：在完成已提交的任务后封闭办事，不再接管新任务；
- shutdownNow()：停止所有正在履行的任务并封闭办事；
- isTerminated()：测试是否所有任务都履行完毕了；
- isShutdown()：测试是否该ExecutorService已被关闭;
- awaitTermination(long,TimeUnit)：接收timeout和TimeUnit两个参数，用于设定超时时间及单位。当等待超过设定时间时，会监测ExecutorService是否已经关闭，若关闭则返回true，否则返回false。一般情况下会和shutdown方法组合使用;
- invokeAll ：作用是等待所有的任务执行完成后统一返回;
- invokeAny ：将第一个得到的结果作为返回值，然后立刻终止所有的线程。如果设置了超时时间，未超时完成则正常返回结果，如果超时未完成则报超时异常。
AbstractExcutorService抽象类
此类的定义并没有特殊的意义仅仅是实现了ExecutorService接口
public abstract class AbstractExecutorService implements ExecutorService {
//此方法很简单就是对runnable保证，将其包装为一个FutureTask
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
//包装callable为FutureTask
//FutureTask其实就是对Callable的一个封装
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
//提交一个Runnable类型的任务
public Future<?> submit(Runnable task) {
//如果为null则抛出NPE
if (task == null) throw new NullPointerException();
//包装任务为一个Future
RunnableFuture<Void> ftask = newTaskFor(task, null);
//将任务丢给执行器，而此处会抛出拒绝异常，在讲述ThreadPoolExecutor的时候有讲述，不记得的读者可以去再看看
execute(ftask);
return ftask;
}
//与上方方法相同只不过指定了返回结果
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
//与上方方法相同只是换成了callable
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
//执行集合tasks结果是最后一个执行结束的任务结果
//可以设置超时 timed为true并且nanos是未来的一个时间
//任何一个任务完成都将会返回结果
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
//传入的任务集合不能为null
if (tasks == null)
throw new NullPointerException();
//传入的任务数不能是0
int ntasks = tasks.size();
if (ntasks == 0)
throw new IllegalArgumentException();
//满足上面的校验后将任务分装到一个ArrayList中
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
//并且创建一个执行器传入this
//这里简单讲述他的执行原理，传入this会使用传入的this(类型为Executor)作为执行器用于执行任务，当submit提交任务的时候回将任务
//封装为一个内部的Future并且重写他的done而此方法就是在future完成的时候调用的，而他的写法则是将当前完成的future添加到esc
//维护的结果队列中
ExecutorCompletionService<T> ecs =
new ExecutorCompletionService<T>(this);
try {
//创建一个执行异常，以便后面抛出
ExecutionException ee = null;
//如果开启了超时则计算死线时间如果时间是0则代表没有开启执行超时
final long deadline = timed ? System.nanoTime() + nanos : 0L;
//获取任务的迭代器
Iterator<? extends Callable<T>> it = tasks.iterator();
//先获取迭代器中的第一个任务提交给前面创建的ecs执行器
futures.add(ecs.submit(it.next()));
//前面记录的任务数减一
--ntasks;
//当前激活数为1
int active = 1;
//进入死循环
for (;;) {
//获取刚才提价的任务是否完成如果完成则f不是null否则为null
Future<T> f = ecs.poll();
//如果为null则代表任务还在继续
if (f == null) {
//如果当前任务大于0 说明除了刚才的任务还有别的任务存在
if (ntasks > 0) {
//则任务数减一
--ntasks;
//并且再次提交新的任务
futures.add(ecs.submit(it.next()));
//当前的存活的执行任务加一
++active;
}
//如果当前存活任务数是0则代表没有任务在执行了从而跳出循环
else if (active == 0)
break;
//如果当前任务执行设置了超时时间
else if (timed) {
//则设置指定的超时时间获取
f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
//等待执行超时还没有获取到则抛出超时异常
if (f == null)
throw new TimeoutException();
//否则使用当前时间计算剩下的超时时间用于下一个循环使用
nanos = deadline - System.nanoTime();
}
//如果没有设置超时则直接获取任务
else
f = ecs.take();
}
//如果获取到了任务结果f!=null
if (f != null) {
//激活数减一
--active;
try {
//返回获取到的结果
return f.get();
//如果获取结果出错则包装异常
} catch (ExecutionException eex) {
ee = eex;
} catch (RuntimeException rex) {
ee = new ExecutionException(rex);
}
}
}
//如果异常不是null则抛出如果是则创建一个
if (ee == null)
ee = new ExecutionException();
throw ee;
} finally {
//其他任务则设置取消
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
//对上方方法的封装
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
try {
return doInvokeAny(tasks, false, 0);
} catch (TimeoutException cannotHappen) {
assert false;
return null;
}
}
//对上方法的封装
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return doInvokeAny(tasks, true, unit.toNanos(timeout));
}
//相对于上一个方法执行成功任何一个则返回结果而此方法是全部执行完然后统一返回结果
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
//传入的任务集合不能是null
if (tasks == null)
throw new NullPointerException();
//创建一个集合用来保存获取到的执行future
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
//任务是否执行完成
boolean done = false;
try {
//遍历传入的任务并且调用执行方法将创建的future添加到集合中
for (Callable<T> t : tasks) {
RunnableFuture<T> f = newTaskFor(t);
futures.add(f);
execute(f);
}
//遍历获取到的future
for (int i = 0, size = futures.size(); i < size; i++) {
Future<T> f = futures.get(i);
//如果当前任务没有成功则进行f.get方法等待此方法执行成功，如果方法执行异常或者被取消将忽略异常
if (!f.isDone()) {
try {
f.get();
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
}
}
}
//到这一步则代表所有的任务都已经有了确切的结果
done = true;
//返回任务结果集合
return futures;
} finally {
//如果不是true是false 则代表执行过程中被中断了则需要对任务进行取消操作，如果正常完成则不会被取消
if (!done)
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
//与上方方法的区别在于对于任务集合可以设置超时时间
//这里会针对差异进行讲解
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
if (tasks == null)
throw new NullPointerException();
//计算设置时长的纳秒时间
long nanos = unit.toNanos(timeout);
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
boolean done = false;
try {
for (Callable<T> t : tasks)
futures.add(newTaskFor(t));
//计算最终计算的确切时间点，运行时长不能超过此时间也就是时间死线
//这里是个细节future创建的时间并没有算作执行时间
final long deadline = System.nanoTime() + nanos;
//获取当前结果数
final int size = futures.size();
//遍历将任务进行执行
for (int i = 0; i < size; i++) {
execute((Runnable)futures.get(i));
//并且每次都计算死线
nanos = deadline - System.nanoTime();
//如果时间已经超过则返回结果
if (nanos <= 0L)
return futures;
}
//否则遍历future确定每次执行都获取到了结果
for (int i = 0; i < size; i++) {
Future<T> f = futures.get(i);
if (!f.isDone()) {
//如果在等待过程中已经超时则返回当前等待结合
if (nanos <= 0L)
return futures;
try {
//如果没有超过死线则设置从future中获取结果的时间如果超过则会派出timeout
f.get(nanos, TimeUnit.NANOSECONDS);
} catch (CancellationException ignore) {
} catch (ExecutionException ignore) {
} catch (TimeoutException toe) {
//抛出了异常则会返回当前的列表
return futures;
}
//计算最新的超时时间
nanos = deadline - System.nanoTime();
}
}
//之前的返回都没有设置为true所以在finally中都会设置为取消唯独正常执行完成到此处返回的结果才是最终的结果
done = true;
return futures;
} finally {
if (!done)
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
}
线程池的具体实现
- ThreadPoolExecutor 默认线程池
- ScheduledThreadPoolExecutor 定时线程池 （下篇再做介绍）
ThreadPoolExecutor
线程池重点属性
//用来标记线程池状态（高3位），线程个数（低29位）
//默认是RUNNING状态，线程个数为0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//线程个数掩码位数，并不是所有平台int类型是32位，所以准确说是具体平台下Integer的二进制位数-3后的剩余位数才是线程的个数，
private static final int COUNT_BITS = Integer.SIZE - 3;
//线程最大个数(低29位)000 11111111111111111111111111111
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
ctl 是对线程池的运行状态和线程池中有效线程的数量进行控制的一个字段， 它包含两部分的信息: 线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount)，这里可以看到，使用了Integer类型来保存，高3位保存runState，低29位保存workerCount。COUNT_BITS 就是29，CAPACITY就是1左移29位减1（29个1），这个常量表示workerCount的上限值，大约是5亿。
ctl相关方法
- runStateOf：获取运行状态；
- workerCountOf：获取活动线程数；
- ctlOf：获取运行状态和活动线程数的值。
// 获取高三位 运行状态
private static int runStateOf(int c) { return c & ~CAPACITY; }
//获取低29位 线程个数
private static int workerCountOf(int c) { return c & CAPACITY; }
//计算ctl新值，线程状态 与 线程个数
private static int ctlOf(int rs, int wc) { return rs | wc; }
线程池存在5种状态
//运行中 111 00000000000000000000000000000
private static final int RUNNING = -1 << COUNT_BITS;
//关闭 000 00000000000000000000000000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
//停止 001 00000000000000000000000000000
private static final int STOP = 1 << COUNT_BITS;
//整理 010 00000000000000000000000000000
private static final int TIDYING = 2 << COUNT_BITS;
//终止 011 00000000000000000000000000000
private static final int TERMINATED = 3 << COUNT_BITS;
使用一个整形,前3位表示状态,后29位表示线程容量,也就是说线程最多有 个
| 前三位 | 状态 | 十进制 |
|---|---|---|
| 111 | RUNNING | -1 |
| 000 | SHUTDOWN | 0 |
| 001 | STOP | 1 |
| 010 | TIDYING | 2 |
| 011 | TERMINATED | 3 |
也可以看出当ctl小于零表示线程池仍在运行
RUNNING
- 状态说明：线程池处在RUNNING状态时，能够接收新任务，以及对已添加的任务进行处理。
- 状态切换：线程池的初始化状态是RUNNING。换句话说，线程池被一旦被创建，就处于RUNNING状态，并且线程池中的任务数为0！
SHUTDOWN
- 状态说明：线程池处在SHUTDOWN状态时，不接收新任务，但能处理已添加的任务。
- 状态切换：调用线程池的shutdown()接口时，线程池由RUNNING -> SHUTDOWN。
STOP
- 状态说明：线程池处在STOP状态时，不接收新任务，不处理已添加的任务，并且会中断正在处理的任务。
- 状态切换：调用线程池的shutdownNow()接口时，线程池由(RUNNING or SHUTDOWN ) -> STOP。
TIDYING
- 状态说明：当所有的任务已终止，ctl记录的”任务数量”为0，线程池会变为TIDYING状态。当线程池变为TIDYING状态时，会执行钩子函数terminated()。terminated()在ThreadPoolExecutor类中是空的，若用户想在线程池变为TIDYING时，进行相应的处理；可以通过重载terminated()函数来实现。
- 状态切换：当线程池在SHUTDOWN状态下，阻塞队列为空并且线程池中执行的任务也为空时，就会由 SHUTDOWN -> TIDYING。 当线程池在STOP状态下，线程池中执行的任务为空时，就会由STOP -> TIDYING。
TERMINATED
- 状态说明：线程池彻底终止，就变成TERMINATED状态。
- 状态切换：线程池处在TIDYING状态时，执行完terminated()之后，就会由 TIDYING -> TERMINATED。
进入TERMINATED的条件如下：
- 线程池不是RUNNING状态；
- 线程池状态不是TIDYING状态或TERMINATED状态；
- 如果线程池状态是SHUTDOWN并且workerQueue为空；
- workerCount为0；
- 设置TIDYING状态成功。
线程池参数
corePoolSize
线程池中的核心线程数，当提交一个任务时，线程池创建一个新线程执行任务，直到当前线程数等于corePoolSize；如果当前线程数为corePoolSize，继续提交的任务被保存到阻塞队列中，等待被执行；如果执行了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有核心线程。
maximumPoolSize
线程池中允许的最大线程数。如果当前阻塞队列满了，且继续提交任务，则创建新的线程执行任务，前提是当前线程数小于maximumPoolSize；
keepAliveTim
线程池维护线程所允许的空闲时间。当线程池中的线程数量大于corePoolSize的时候，如果这时没有新的任务提交，核心线程外的线程不会立即销毁，而是会等待，直到等待的时间超过了keepAliveTime；
unit
keepAliveTime的单位；
workQueue
用来保存等待被执行的任务的阻塞队列，且任务必须实现Runable接口，在JDK中提供了如下阻塞队列：
- 1、ArrayBlockingQueue：基于数组结构的有界阻塞队列，按FIFO排序任务；
- 2、LinkedBlockingQuene：基于链表结构的阻塞队列，按FIFO排序任务，吞吐量通常要高于ArrayBlockingQuene；
- 3、SynchronousQuene：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQuene；
- 4、priorityBlockingQuene：具有优先级的无界阻塞队列；
threadFactory
它是ThreadFactory类型的变量，用来创建新线程。默认使用Executors.defaultThreadFactory() 来创建线程。使用默认的ThreadFactory来创建线程时，会使新创建的线程具有相同的NORM_PRIORITY优先级并且是非守护线程，同时也设置了线程的名称。
handler
线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，线程池提供了4种策略：
- AbortPolicy：直接抛出异常，默认策略；
- CallerRunsPolicy：用调用者所在的线程来执行任务；
- DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
- DiscardPolicy：直接丢弃任务；
上面的4种策略都是ThreadPoolExecutor的内部类。
当然也可以根据应用场景实现RejectedExecutionHandler接口，自定义饱和策略，如记录日志或持久化存储不能处理的任务。
线程池的创建
有四个构造函数，其他三个都是调用下面代码中的这个构造函数
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
线程池监控
public long getTaskCount() //线程池已执行与未执行的任务总数
public long getCompletedTaskCount() //已完成的任务数
public int getPoolSize() //线程池当前的线程数
public int getActiveCount() //线程池中正在执行任务的线程数量
线程池原理
核心方法分析
execute方法
execute 方法是提交任务 command 到线程池进行执行
public void execute(Runnable command) {
//如果任务为null，则抛出NPE异常
if (command == null)
throw new NullPointerException();
/*
* clt记录着runState和workerCount
*/
int c = ctl.get();
/*
* workerCountOf方法取出低29位的值，表示当前活动的线程数；
* 如果当前活动线程数小于corePoolSize，则新建一个线程放入线程池中；
* 并把任务添加到该线程中。
*/
if (workerCountOf(c) < corePoolSize) {
/*
* addWorker中的第二个参数表示限制添加线程的数量是根据corePoolSize来判断还是maximumPoolSize来判断；
* 如果为true，根据corePoolSize来判断；
* 如果为false，则根据maximumPoolSize来判断
*/
if (addWorker(command, true))
return;
/*
* 如果添加失败，则重新获取ctl值
*/
c = ctl.get();
}
/*
* 如果当前线程池是运行状态并且任务添加到队列成功
*/
if (isRunning(c) && workQueue.offer(command)) {
// 重新获取ctl值
int recheck = ctl.get();
// 再次判断线程池的运行状态，如果不是运行状态，由于之前已经把command添加到workQueue中了，
// 这时需要移除该command
// 执行过后通过handler使用拒绝策略对该任务进行处理，整个方法返回
if (! isRunning(recheck) && remove(command))
reject(command);
/*
* 获取线程池中的有效线程数，如果数量是0，则执行addWorker方法
* 这里传入的参数表示：
* 1. 第一个参数为null，表示在线程池中创建一个线程，但不去启动；
* 2. 第二个参数为false，将线程池的有限线程数量的上限设置为maximumPoolSize，添加线程时根据maximumPoolSize来判断；
* 如果判断workerCount大于0，则直接返回，在workQueue中新增的command会在将来的某个时刻被执行。
*/
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
/*
* 如果执行到这里，有两种情况：
* 1. 线程池已经不是RUNNING状态；
* 2. 线程池是RUNNING状态，但workerCount >= corePoolSize并且workQueue已满。
* 这时，再次调用addWorker方法，但第二个参数传入为false，将线程池的有限线程数量的上限设置为maximumPoolSize；
* 如果失败则拒绝该任务
*/
else if (!addWorker(command, false))
reject(command);
}
简单来说，在执行execute()方法时如果状态一直是RUNNING时的执行过程如下：
- 如果 workerCount < corePoolSize，则创建并启动一个线程来执行新提交的任务；
- 如果 workerCount > corePoolSize，且线程池内的阻塞队列未满，则将任务添加到阻塞队列中；
- 如果 workerCount >= corePoolSize && workerCount < maximumPoolSize，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务；
- 如果 workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满，则根据拒绝策略来处理该任务，默认的处理方式是直接抛异常。
注意：addWorker(null,false); ，也就是创建一个线程，但并没有传入从任务，因为任务已经被添加到 workQueue中了，所以 worker 在执行的时候，会直接从 workQueue 中获取任务。所以，在workerCountof(recheck) == 0 时执行 addWorker(null,false); 也是为了保证线程池在RUNNING状态下必须要有一个线程来执行任务。
execute方法执行流程如下：
addWorker方法
addWorker方法的主要工作是在线程池中创建一个新的线程并执行：
- firstTask参数 用于指定新增的线程执行的第一个任务；
- core参数 为true表示在新增线程时会判断当前活动线程数是否少于corePoolSize，false表示新增线程前需要判断当前活动线程数是否少于maximumPoolSize。
代码如下：
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
//获取ctl
int c = ctl.get();
// 获取运行状态
int rs = runStateOf(c);
/*
* 这个if判断
* 如果rs >= SHUTDOWN，则表示此时不再接收新任务；
* 接着判断以下3个条件，只要有1个不满足，则返回false：
* 1. rs == SHUTDOWN，这时表示关闭状态，不再接受新提交的任务，但却可以继续处理阻塞队列中已保存的任务
* 2. firsTask为空
* 3. 阻塞队列不为空
*
* 首先考虑rs == SHUTDOWN的情况
* 这种情况下不会接受新提交的任务，所以在firstTask不为空的时候会返回false；
* 然后，如果firstTask为空，并且workQueue也为空，则返回false，
* 因为队列中已经没有任务了，不需要再添加线程了
*/
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
// 获取线程数
int wc = workerCountOf(c);
// 如果wc超过CAPACITY，也就是ctl的低29位的最大值（二进制是29个1），返回false；
// 这里的core是addWorker方法的第二个参数，如果为true表示根据corePoolSize来比较，
// 如果为false则根据maximumPoolSize来比较。
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 尝试增加workerCount，如果成功，则跳出第一个for循环
if (compareAndIncrementWorkerCount(c))
break retry;
// 如果增加workerCount失败，则重新获取ctl的值
c = ctl.get(); // Re-read ctl
// 如果当前的运行状态不等于rs，说明状态已被改变，返回第一个for循环继续执行
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 {
// 根据firstTask来创建Worker对象
w = new Worker(firstTask);
// 每一个Worker对象都会创建一个线程
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
// rs < SHUTDOWN表示是RUNNING状态；
// 如果rs是RUNNING状态或者rs是SHUTDOWN状态并且firstTask为null，向线程池中添加线程。
// 因为在SHUTDOWN时不会在添加新的任务，但还是会执行workQueue中的任务
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// workers是一个HashSet
workers.add(w);
int s = workers.size();
// largestPoolSize记录着线程池中出现过的最大线程数量
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 启动线程
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
Worker类
线程池中的每一个线程被封装成一个 Worker 对象，ThreadPool 维护的其实就是一组 Worker 对象。
/**
* Woker主要维护着运行task的worker的中断控制信息，以及其他小记录。这个类继承AbstractQueuedSynchronizer
* 而来简化获取和释放每一个任务执行中的锁。这可以防止中断那些打算唤醒正在等待其他线程任务的任务，而不是
* 中断正在运行的任务。我们实现一个简单的不可重入锁而不是ReentrantLock，因为我们不想当其调用setCorePoolSize
* 这样的方法的时候能获得锁。
*/
//worker主要是对进行中的任务进行中断控制，顺带着对其他行为进行记录
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
//正在跑的线程，如果是null标识factory失败
final Thread thread;
//初始化一个任务以运行
Runnable firstTask;
//每个线程计数
volatile long completedTasks;
/**
* 用给定的firstTask和从threadFactory创建
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
//主要调用了runWorker
public void run() {
runWorker(this);
}
// Lock methods
//
// 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类继承了AQS，并实现了Runnable接口，注意其中的 firstTask 和 thread 属性：
- firstTask用它来保存传入的任务；
- thread是在调用构造方法时通过 ThreadFactory 来创建的线程，是用来处理任务的线程。
在调用构造方法时，需要把任务传入，这里通过 getThreadFactory().newThread(this); 来新建一个线程，newThread 方法传入的参数是 this，因为 Worker 本身继承了 Runnable 接口，也就是一个线程，所以一个 Worker 对象在启动的时候会调用 Worker 类中的 run 方法。
Worker 继承了 AQS，使用AQS来实现独占锁的功能。为什么不使用 ReentrantLock 来实现呢？可以看到 tryAcquire 方法，它是不允许重入的，而 ReentrantLock 是允许重入的：
- lock方法一旦获取了独占锁，表示当前线程正在执行任务中；
- 如果正在执行任务，则不应该中断线程；
- 如果该线程现在不是独占锁的状态，也就是空闲的状态，说明它没有在处理任务，这时可以对该线程进行中断；
- 线程池在执行 shutdown 方法或 tryTerminate 方法时会调用 interruptIdleWorkers 方法来中断空闲的线程，interruptIdleWorkers 方法会使用 tryLock 方法来判断线程池中的线程是否是空闲状态；
- 之所以设置为不可重入，是因为我们不希望任务在调用像 setCorePoolSize 这样的线程池控制方法时重新获取锁。如果使用ReentrantLock，它是可重入的，这样如果在任务中调用了如 setCorePoolSize 这类线程池控制的方法，会中断正在运行的线程。
所以，Worker继承自AQS，用于判断线程是否空闲以及是否可以被中断。
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
此外，在构造方法中执行了 setState(-1); ，把 state 变量设置为 -1，为什么这样做呢？是因为AQS中默认的 state 是0，如果刚刚创建了一个 Worker 对象，还没有执行任务时，这时就不应该被中断，看一下 tryAquire 方法：
protected boolean tryAcquire(int unused) {
//cas修改state，不可重入
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
tryAcquire方法是根据state是否是0来判断的，所以，setState(-1);将state设置为-1是为了禁止在执行任务前对线程进行中断。
既然是AQS,其对于state的解释也很关键
| state | 含义 |
|---|---|
| -1 | 初始化状态,禁止中断 |
| 0 | 解锁状态 |
| 1 | 上锁状态(独占) |
正因为如此，在runWorker方法中会先调用Worker对象的unlock方法将state设置为0。
runWorker方法
在Worker类中的run方法调用了runWorker方法来执行任务，runWorker方法的代码如下：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 获取第一个任务
Runnable task = w.firstTask;
w.firstTask = null;
// 解锁,允许interrupt操作
w.unlock(); // allow interrupts
// 是否因为异常退出循环
boolean completedAbruptly = true;
try {
// 如果task为空，则通过getTask来获取任务
while (task != null || (task = getTask()) != null) {
// 这儿对worker进行加锁，是为了达到下面的目的
// 1. 降低锁范围，提升性能
// 2. 保证每个worker执行的任务是串行的
w.lock();
// 如果线程池处于STOP状态就中断线程
// 如果线程被中断(清除中断标记),线程池处于STOP状态,线程没有再被中断
if ((runStateAtLeast(ctl.get(), STOP) || //至少是STOP状态
(Thread.interrupted() && //中断过(并抹除中断标记)
runStateAtLeast(ctl.get(), STOP))) && //再次检查
!wt.isInterrupted()) //wt没有被中断
wt.interrupt();
try {
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 {
afterExecute(task, thrown); //钩子方法
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
这里说明一下第一个 if判断 ，目的是：
- 如果线程池正在停止，那么要保证当前线程是中断状态；
- 如果不是的话，则要保证当前线程不是中断状态；
这里要考虑在执行该 if语句 期间可能也执行了 shutdownNow 方法，shutdownNow 方法会把状态设置为 STOP ，回顾一下 STOP 状态：不能接受新任务，也不处理队列中的任务，会中断正在处理任务的线程。在线程池处于 RUNNING 或 SHUTDOWN 状态时，调用 shutdownNow() 方法会使线程池进入该状态。
总结一下runWorker方法的执行过程：
- while循环不断地通过getTask()方法获取任务；
- getTask()方法从阻塞队列中取任务；
- 如果线程池正在停止，那么要保证当前线程是中断状态，否则要保证当前线程不是中断状态；
- 调用task.run()执行任务；
- 如果task为null则跳出循环，执行processWorkerExit()方法；
- runWorker方法执行完毕，也代表着Worker中的run方法执行完毕，销毁线程。
这里的beforeExecute方法和afterExecute方法在ThreadPoolExecutor类中是空的，留给子类来实现。
completedAbruptly 变量来表示执行任务过程中是否出现了异常，在processWorkerExit方法中会对该变量的值进行判断。
getTask方法
getTask方法用来从阻塞队列中取任务，代码如下：
private Runnable getTask() {
// timeOut变量的值表示上次从阻塞队列中取任务时是否超时
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
/*
* 如果线程池状态rs >= SHUTDOWN，也就是非RUNNING状态，再进行以下判断：
* 1. rs >= STOP，线程池是否正在stop；
* 2. 阻塞队列是否为空。
* 如果以上条件满足，则将workerCount减1并返回null。
* 因为如果当前线程池状态的值是SHUTDOWN或以上时，不允许再向阻塞队列中添加任务。
*/
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
// timed变量用于判断是否需要进行超时控制。
// allowCoreThreadTimeOut默认是false，也就是核心线程不允许进行超时；
// wc > corePoolSize，表示当前线程池中的线程数量大于核心线程数量；
// 对于超过核心线程数量的这些线程，需要进行超时控制
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
/*
* wc > maximumPoolSize的情况是因为可能在此方法执行阶段同时执行了setMaximumPoolSize方法；
* timed && timedOut 如果为true，表示当前操作需要进行超时控制，并且上次从阻塞队列中获取任务发生了超时
* 接下来判断，如果有效线程数量大于1，或者阻塞队列是空的，那么尝试将workerCount减1；
* 如果减1失败，则返回重试。
* 如果wc == 1时，也就说明当前线程是线程池中唯一的一个线程了。
*/
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
/*
* 根据timed来判断，如果为true，则通过阻塞队列的poll方法进行超时控制，如果在keepAliveTime时间内没有获取到任务，则返回null；
* 否则通过take方法，如果这时队列为空，则take方法会阻塞直到队列不为空。
*
*/
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
// 如果 r == null，说明已经超时，timedOut设置为true
timedOut = true;
} catch (InterruptedException retry) {
// 如果获取任务时当前线程发生了中断，则设置timedOut为false并返回循环重试
timedOut = false;
}
}
}
这里重要的地方是第二个 if判断 ，目的是控制线程池的有效线程数量。由上文中的分析可以知道，在执行 execute 方法时，如果当前线程池的线程数量超过了 corePoolSize 且小于 maximumPoolSize，并且workQueue已满时，则可以添加工作线程，但这时如果超时没有获取到任务，也就是 timedOut 为 true的情况，说明 workQueue 已经为空了，也就说明了当前线程池中不需要那么多线程来执行任务了，可以把多余 corePoolSize 数量的线程销毁掉，保持线程数量在 corePoolSize 即可。
什么时候会销毁？
当然是runWorker方法执行完之后，也就是Worker中的run方法执行完，由JVM自动回收。
getTask方法返回null时，在runWorker方法中会跳出while循环，然后会执行 processWorkerExit方法。
processWorkerExit方法
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 如果completedAbruptly值为true，则说明线程执行时出现了异常，需要将workerCount减1；
// 如果线程执行时没有出现异常，说明在getTask()方法中已经已经对workerCount进行了减1操作，这里就不必再减了。
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//统计完成的任务数
completedTaskCount += w.completedTasks;
// 从workers中移除，也就表示着从线程池中移除了一个工作线程
workers.remove(w);
} finally {
mainLock.unlock();
}
// 根据线程池状态进行判断是否结束线程池
tryTerminate();
int c = ctl.get();
/*
* 当线程池是RUNNING或SHUTDOWN状态时，如果worker是异常结束，那么会直接addWorker；
* 如果allowCoreThreadTimeOut=true，并且等待队列有任务，至少保留一个worker；
* 如果allowCoreThreadTimeOut=false，workerCount不少于corePoolSize。
*/
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);
}
}
至此，processWorkerExit执行完成之后，工作线程被销毁，以上就是整个工作线程的生命周期，从 execute 方法开始，Worker 使用 ThreadFactory 创建新的工作线程，runWorker 通过 getTask 获取任务，然后执行任务，如果 getTask 返回null，进入 oricessWorkerExit 方法，整个线程结束。如下图所示：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-aOmgFHaU-1607357995983)(https://imgkr2.cn-bj.ufileos.com/5171fa57-6556-4adb-a6e3-98a1805792e1.jpg?UCloudPublicKey=TOKEN_8d8b72be-579a-4e83-bfd0-5f6ce1546f13&Signature=d%252Fo2TdoPfih5GlF0LZOsql0wfWQ%253D&Expires=1607443308)]
shutdown方法
调用关系
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-sfdnAYMr-1607357995986)(https://imgkr2.cn-bj.ufileos.com/cc964937-ee9d-4c8b-8d99-68b05c96c158.jpg?UCloudPublicKey=TOKEN_8d8b72be-579a-4e83-bfd0-5f6ce1546f13&Signature=ML48OwlgyVP5kZwQPCm99qp%252FGoU%253D&Expires=1607443292)]
shutdown
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();//权限检查
advanceRunState(SHUTDOWN);//设置当前线程池状态为SHUTDOWN
interruptIdleWorkers();//中断空闲线程
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
interruptIdleWorkers
中断空闲线程
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();//加锁
try {
for (Worker w : workers) {
Thread t = w.thread;
if (!t.isInterrupted() && w.tryLock()) {//未被中断且正在独占运行
try {
t.interrupt();//中断
} catch (SecurityException ignore) {
} finally {
w.unlock();//worker解锁
}
}
if (onlyOne)//如果只中断一个
break;
}
} finally {
mainLock.unlock();//解锁
}
}
tryTerminate
final void tryTerminate() {
for (;;) { // 无限循环，确保操作成功
// 获取线程池控制状态
int c = ctl.get();
if (isRunning(c) || // 线程池的运行状态为RUNNING
runStateAtLeast(c, TIDYING) || // 线程池的运行状态最小要大于TIDYING
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) // 线程池的运行状态为SHUTDOWN并且workQueue队列不为null
// 不能终止，直接返回
return;
if (workerCountOf(c) != 0) { // 线程池正在运行的worker数量不为0 // Eligible to terminate
// 仅仅中断一个空闲的worker
interruptIdleWorkers(ONLY_ONE);
return;
}
// 获取线程池的锁
final ReentrantLock mainLock = this.mainLock;
// 获取锁
mainLock.lock();
try {
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { // 比较并设置线程池控制状态为TIDYING
try {
// 终止，钩子函数
terminated();
} finally {
// 设置线程池控制状态为TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
// 释放在termination条件上等待的所有线程
termination.signalAll();
}
return;
}
} finally {
// 释放锁
mainLock.unlock();
}
// else retry on failed CAS
}
}
shutdownNow方法
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);//设置线程池状态为STOP
interruptWorkers();//中断线程
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
Executors类
Executors类，提供了一系列工厂方法用于创建线程池，返回的线程池都实现了ExecutorService接口。
关于Callable的支持
- Callable<Object> callable(Runnable task)
返回 Callable 对象，调用它时可运行给定的任务。
public static Callable<Object> callable(Runnable task) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<Object>(task, null);
}
- Callable<T> callable(Runnable task, T result)
返回 Callable 对象，调用它时可运行给定的任务并返回给定的结果。callable(task)等价于callable(task, null)。
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
RunnableAdapter类
/**
* A callable that runs given task and returns given result
*/
static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}
}
- Callable callable(final PrivilegedAction action)
返回 Callable 对象，调用它时可运行给定特权的操作并返回其结果。
public static Callable<Object> callable(final PrivilegedAction<?> action) {
if (action == null)
throw new NullPointerException();
return new Callable<Object>() {
public Object call() { return action.run(); }
};
}
- callable(PrivilegedExceptionAction action)
和Callable callable(final PrivilegedAction action)其类似，唯一的区别在于，前者可抛出异常。
public static Callable<Object> callable(final PrivilegedExceptionAction<?> action) {
if (action == null)
throw new NullPointerException();
return new Callable<Object>() {
public Object call() throws Exception { return action.run(); }
};
}
PrivilegedAction接口
public interface PrivilegedAction<T> {
T run();
}
- Callable privilegedCallable(Callable<T> callable)
返回 Callable 对象，调用它时可在当前的访问控制上下文中执行给定的 callable 对象。
public static <T> Callable<T> privilegedCallable(Callable<T> callable) {
if (callable == null)
throw new NullPointerException();
return new PrivilegedCallable<T>(callable);
}
- Callable privilegedCallableUsingCurrentClassLoader(Callable<T> callable)
返回 Callable 对象，调用它时可在当前的访问控制上下文中，使用当前上下文类加载器作为上下文类加载器来执行给定的 callable 对象。
public static <T> Callable<T> privilegedCallableUsingCurrentClassLoader(Callable<T> callable) {
if (callable == null)
throw new NullPointerException();
return new PrivilegedCallableUsingCurrentClassLoader<T>(callable);
}
关于ThreadFactory的支持
- 根据需要创建新线程的对象。使用线程工厂就无需再手工编写对 new Thread 的调用了，从而允许应用程序使用特殊的线程子类、属性等等。
- Executors对其提供支持：DefaultThreadFactory和PrivilegedThreadFactory。
public interface ThreadFactory {
Thread newThread(Runnable r);
}
ThreadFactory为默认的defaultThreadFactory，ThreadFactory定义了两个：defaultThreadFactory和privilegedThreadFactory，但是也可以自己定义，实现ThreadFactory接口或者继承已经实现的这两个实现并且修改对应的代码即可。
// 返回用于创建新线程的默认线程工厂。
public static ThreadFactory defaultThreadFactory() {
return new DefaultThreadFactory();
}
// 返回用于创建新线程的线程工厂，这些新线程与当前线程具有相同的权限。
public static ThreadFactory privilegedThreadFactory() {
return new PrivilegedThreadFactory();
}
- DefaultThreadFactory
static class DefaultThreadFactory implements ThreadFactory {
static final AtomicInteger poolNumber = new AtomicInteger(1);
final ThreadGroup group;
final AtomicInteger threadNumber = new AtomicInteger(1);
final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null)? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
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;
}
}
从源码看出DefaultThreadFactory就是创建一个普通的线程，非守护线程，优先级为5。
新线程具有可通过 pool-N-thread-M 的 Thread.getName() 来访问的名称，其中 N 是此工厂的序列号， M 是此工厂所创建线程的序列号。
- PrivilegedThreadFactory
static class PrivilegedThreadFactory extends DefaultThreadFactory {
private final ClassLoader ccl;
private final AccessControlContext acc;
PrivilegedThreadFactory() {
super();
this.ccl = Thread.currentThread().getContextClassLoader();
this.acc = AccessController.getContext();
acc.checkPermission(new RuntimePermission("setContextClassLoader"));
}
public Thread newThread(final Runnable r) {
return super.newThread(new Runnable() {
public void run() {
AccessController.doPrivileged(new PrivilegedAction<Object>() {
public Object run() {
Thread.currentThread().setContextClassLoader(ccl);
r.run();
return null;
}
}, acc);
}
});
}
}
从源码看出，PrivilegedThreadFactory extends DefaultThreadFactory从而具有与 defaultThreadFactory() 相同设置的线程。但增加了两个特性：ClassLoader和AccessControlContext，从而使运行在此类线程中的任务具有与当前线程相同的访问控制和类加载器。
关于RejectedExecutionHandler的支持
而默认的拒绝策略RejectedExecutionHandler则为 AbortPolicy，并且所有的拒绝策略都是实现接口RejectedExecutionHandler
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
五种线程池创建类型
newFixedThreadPool（固定大小线程池）
newFixedThreadPool：创建一个核心线程个数和最大线程个数都为 nThreads 的线程池，并且阻塞队列长度为 Integer.MAX_VALUE，keeyAliveTime=0 说明只要线程个数比核心线程个数多并且当前空闲则回收。代码如下：
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
//使用自定义线程创建工厂
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory);
}
newSingleThreadExecutor(单个后台线程）
newSingleThreadExecutor：创建一个核心线程个数和最大线程个数都为1的线程池，并且阻塞队列长度为 Integer.MAX_VALUE，keeyAliveTime=0 说明只要线程个数比核心线程个数多并且当前空闲则回收。代码如下：
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
//使用自己的线程工厂
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory));
}
newCachedThreadPool（无界线程池，可以进行自动线程回收）
newCachedThreadPool：创建一个按需创建线程的线程池，初始线程个数为 0，最多线程个数为 Integer.MAX_VALUE，并且阻塞队列为同步队列，keeyAliveTime=60 说明只要当前线程 60s 内空闲则回收。这个特殊在于加入到同步队列的任务会被马上被执行，同步队列里面最多只有一个任务。代码如下：
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
//使用自定义的线程工厂
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
threadFactory);
}
newScheduledThreadPool (可调度)
newScheduledThreadPool：创建一个定长线程池，支持定时及周期性任务执行。newScheduledThreadPool 和 其他线程池最大的区别是使用的阻塞队列是 DelayedWorkQueue，而且多了两个定时执行的方法scheduleAtFixedRate和scheduleWithFixedDelay
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
//使用自定义的线程工厂
public static ScheduledExecutorService newScheduledThreadPool(
int corePoolSize, ThreadFactory threadFactory) {
return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
newWorkStealingPool（并行操作）
newWorkStealingPool：JDK1.8新增newWorkStealingPool，适合使用在很耗时的操作，但是newWorkStealingPool不是ThreadPoolExecutor的扩展，它是新的线程池类ForkJoinPool的扩展，但是都是在统一的一个Executors类中实现，由于能够合理的使用CPU进行对任务操作（并行操作），所以适合使用在很耗时的任务中。代码如下：
public static ExecutorService newWorkStealingPool() {
return new ForkJoinPool
(Runtime.getRuntime().availableProcessors(),
ForkJoinPool.defaultForkJoinWorkerThreadFactory,
null, true);
}
public static ExecutorService newWorkStealingPool(int parallelism) {
return new ForkJoinPool
(parallelism,
ForkJoinPool.defaultForkJoinWorkerThreadFactory,
null, true);
}
相关文章
- 并发编程从操作系统底层工作的整体认识开始
- 深入理解Java内存模型(JMM)及volatile关键字
- 深入理解CPU缓存一致性协议(MESI)
- 并发编程之synchronized深入理解
- 并发编程之抽象队列同步器AQS应用Lock详解
- 阻塞队列 — ArrayBlockingQueue源码分析
- 阻塞队列 — LinkedBlockingQueue源码分析
- 阻塞队列 —PriorityBlockingQueue源码分析
- 阻塞队列 — DelayQueue源码分析
- 阻塞队列 — SynchronousQueue源码分析
- 阻塞队列 — LinkedTransferQueue源码分析
- 阻塞队列 — LinkedBlockingDeque源码分析
- 阻塞队列 — DelayedWorkQueue源码分析
- 并发编程之Future&FutureTask深入解析
PS：以上代码提交在 Github ：https://github.com/Niuh-Study/niuh-juc-final.git
文章持续更新，可以公众号搜一搜「 一角钱技术 」第一时间阅读， 本文 GitHub org_hejianhui/JavaStudy 已经收录，欢迎 Star。
