一、说明
ThreadPoolExecutor
- 继承
Executor接口 - 它有多个构造方法来实现自定义创建线程池，以内部线程池的形式对外提供管理任务执行，线程调度，线程池管理等
- 关闭线程池调用
shutdown()、shutdownNow()、awaitTermination()方法
二、理解
shutdown()
- 只关闭了提交通道，停止接收新任务，已提交的任务会继续执行直到完成，此方法不会阻塞，当所有提交任务执行完毕，线程池被关闭
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
shutdownNow()
- 立即停止线程池，停止接收新任务，中断所有正在执行的任务，停止对等待队列的处理，立刻返回未执行的任务列表
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
interruptWorkers();
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
awaitTermination()
- 此方法阻塞，在
shutdown()调用之后，等待所有已提交的任务完成或者超时，在这之后可以继续提交任务 - 如果
ExecutorService在超时之前已经关闭，返回true，即所有任务执行完毕；否则返回false，即已超时 - 如果不在之前调用
shutdown()，即使在超时之前所有任务执行完成，ExecutorService仍未终止，awaitTermination()依然在等待终止状态，返回false
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
while (!runStateAtLeast(ctl.get(), TERMINATED)) {
if (nanos <= 0L)
return false;
nanos = termination.awaitNanos(nanos);
}
return true;
} finally {
mainLock.unlock();
}
}
线程状态转换关系
runState
- 线程池的生命周期
- 线程池的状态(runState)和工作线程数量(workerCount)共同保存在 AtomicInteger 类型的控制变量 ctl 中
- AtomicInteger 类对 int 类型进行了封装，可以对整数进行各种原子操作
- ctl高三位保存运行状态(23=8>5)，低29位保存工作线程的数量(229-1)
// 初始运行状态为RUNNING，线程数为0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// COUNT_BITS: 29
private static final int COUNT_BITS = Integer.SIZE - 3;
// CAPACITY: 十进制: 536870911 二进制: 00011111111111111111111111111111
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
// RUNNING: 十进制:-536870912 二进制:11100000000000000000000000000000
private static final int RUNNING = -1 << COUNT_BITS;
// SHUTDOWN: 十进制:0 二进制:0
private static final int SHUTDOWN = 0 << COUNT_BITS;
// STOP: 十进制:536870912 二进制:00100000000000000000000000000000
private static final int STOP = 1 << COUNT_BITS;
// TIDYING: 十进制:1073741824 二进制:01000000000000000000000000000000
private static final int TIDYING = 2 << COUNT_BITS;
// TERMINATED: 十进制:1610612736 二进制:01100000000000000000000000000000
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl 打包和解包ctl
// 获取线程池当前状态，CAPACITY取反，高三位都是1，低29位都是0，和ctl进行与运算，获得runState变量
private static int runStateOf(int c) { return c & ~CAPACITY; }
// CAPACITY高三位都是0，低29位都是0，和ctl进行与运算获得workerCount变量
private static int workerCountOf(int c) { return c & CAPACITY; }
// 初始化ctl变量，runState和workerCount进行或运算后共同存储在一个变量中
private static int ctlOf(int rs, int wc) { return rs | wc; }
RUNNING接收新的任务，并且可执行队列里的任务SHUTDOWN停止接收新任务，但可执行队列里的任务STOP可执行队列里的任务，不执行队列里的任务，中断正在执行的任务TIDYING所有任务都已终止，线程数为0，当线程池变为TIDYING状态时，会执行钩子函数terminated()，钩子方法是指使用一个抽象类实现接口，
一个抽象类实现这个接口，需要的方法设置为abstract，其它设置为空方法TERMINATED终止状态，表示线程池已关闭，已经执行完terminated()钩子方法
判断当前线程池运行状态
// 判断线程池当前运行状态是否小于给定值
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
// 判断线程池当前运行状态是否大于等于给定值
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
// 判断线程池是否处于RUNNING状态
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
// 判断线程池是否处于SHUTDOWN状态
public boolean isShutdown() {
return ! isRunning(ctl.get());
}
// 判断线程池是否处于TERMINATING状态
public boolean isTerminating() {
int c = ctl.get();
return ! isRunning(c) && runStateLessThan(c, TERMINATED);
}
// 判断线程池是否处于TERMINATED状态
public boolean isTerminated() {
return runStateAtLeast(ctl.get(), TERMINATED);
}
运行状态转换关系
三、实现
1.shutdown()
创建一个ShutdownTest类，默认使用ThreadPoolExecutor.AbortPolicy拒绝策略，队列是ArrayBlockingQueue，设置核心线程数最大值为1，线程池线程数最大值为2，最大等待时间为5秒，等待队列值为2，提交8个任务，在第5个任务的时候执行 shutdown()
public class ShutdownTest {
public static void main(String[] args) {
// 1.创建线程池
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 2, 5,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),
Executors.defaultThreadFactory());
// 2.创建线程任务
for (int i = 1; i <= 8; i++) {
// 3.执行任务
System.out.println("执行第"+i+"个任务");
threadPoolExecutor.execute(new runnable("任务"+i));
// 4.获取等待队列
Iterator iterator = threadPoolExecutor.getQueue().iterator();
System.out.print("当前等待队列 ");
while (iterator.hasNext()){
runnable thread = (runnable) iterator.next();
System.out.print(thread.name + "\t");
}
System.out.print("\n");
System.out.println("--------");
// 5.关闭线程池
if (i == 4) {
threadPoolExecutor.shutdown();
System.out.println("线程池已关闭");
}
}
}
static class runnable implements Runnable{
// 设置任务名
String name;
public runnable(String setName) {
this.name = setName;
}
@Override
public void run() {
try {
System.out.println("线程:"+Thread.currentThread().getName() +" 执行: "+name);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
执行 shutdown()，此时停止接收新任务，已提交的任务会继续执行直到完成，此方法不会阻塞，抛出RejectedExecutionException
如果捕获RejectedExecutionException，可以看到任务被拒绝了
public static void main(String[] args) {
// 1.创建线程池
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 2, 5,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),
Executors.defaultThreadFactory());
// 2.创建线程任务
for (int i = 1; i <= 8; i++) {
// 3.执行任务
System.out.println("执行第"+i+"个任务");
try {
threadPoolExecutor.execute(new runnable("任务"+i));
// 4.获取等待队列
Iterator iterator = threadPoolExecutor.getQueue().iterator();
System.out.print("当前等待队列 ");
while (iterator.hasNext()){
runnable thread = (runnable) iterator.next();
System.out.print(thread.name + "\t");
}
System.out.print("\n");
System.out.println("--------");
} catch (RejectedExecutionException e) {
// 5.捕获拒绝执行策略异常
System.out.println("拒绝执行第" + i + "个任务");
}
// 6.关闭线程池
if (i == 4) {
threadPoolExecutor.shutdown();
System.out.println("线程池已关闭");
}
}
}
2.shutdownNow()
立即停止线程池，停止接收新任务，中断所有正在执行的任务，停止对等待队列的处理
// 6.关闭线程池
if (i == 4) {
threadPoolExecutor.shutdownNow();
System.out.println("线程池已关闭");
}
3.awaitTermination()
此方法阻塞，在shutdown()调用之后，等待所有已提交的任务完成或者超时，在这之后可以继续提交任务
如果 shutdown 在 awaitTermination后调用的话，在awaitTermination 未超时前，它不会释放锁；而 shutdown 也无法得到锁去让线程池停止。这就形成了死锁
