ThreadPoolExecutor
Executor 接口shutdown() 、shutdownNow()、awaitTermination()方法shutdown()
public void shutdown() {
// 上锁确保只有一个线程执行此操作
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 检查是否有权限关闭线程池以和中断线程
checkShutdownAccess();
// 将线程池状态设置为SHUTDOWN
advanceRunState(SHUTDOWN);
// 中断所有空闲线程
interruptIdleWorkers();
// 用于取消延时任务
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
// 将线程池置为TERMINATED状态
tryTerminate();
}
shutdownNow()
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
// 上锁确保只有一个线程执行此操作
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 检查是否有权限关闭线程池以和中断线程
checkShutdownAccess();
// 将线程池运行状态置为STOP
advanceRunState(STOP);
// 中断所有线程
interruptWorkers();
// 将未执行的任务移入列表中
tasks = drainQueue();
} finally {
mainLock.unlock();
}
// 将线程池置为TERMINATED状态
tryTerminate();
return tasks;
}
awaitTermination()
在shutdown()调用之后使用，阻塞当前线程，在这之后可以继续提交任务，设置等待超时时间，等待所有任务都执行完成，检查线程池是否终止，如果终止返回 true，否则返回 false，并解除阻塞
如果在超时之前所有任务执行完毕，表示线程池已经终止，返回true，否则返回false
如果在shutdown()之前使用，线程池未终止，awaitTermination()锁在等待终止状态，造成死锁
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
// 将时间单位转化为纳秒
long nanos = unit.toNanos(timeout);
// 上锁确保只有一个线程执行此操作
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 如果线程池在超池之前终止，返回true
while (!runStateAtLeast(ctl.get(), TERMINATED)) {
// 超时但是线程池未终止，返回false
if (nanos <= 0L)
return false;
// 实现阻塞
nanos = termination.awaitNanos(nanos);
}
return true;
} finally {
mainLock.unlock();
}
}
线程池的生命周期
AtomicInteger 类型的控制变量 ctl 中 // 初始运行状态为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);
}
运行状态转换关系
创建一个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("线程池已关闭");
}
}
}
立即停止线程池，停止接收新任务，中断所有正在执行的任务，停止对等待队列的处理
// 6.关闭线程池
if (i == 4) {
threadPoolExecutor.shutdownNow();
System.out.println("线程池已关闭");
}
此方法阻塞，在shutdown()调用之后，停止接收新任务，但是awaitTermination()后可以继续提交，此方法是阻塞的，用来检测timeout时间后线程池是否终止，如果停止，则返回true并释放锁
public class ShutdownTest {
public static void main(String[] args) throws InterruptedException {
// 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();
while (!threadPoolExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
System.out.println("线程池未关闭");
}
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);
Thread.sleep(5000);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
保证了线程池和其他线程的之间的执行顺序
如果 shutdown() 在 awaitTermination()后调用的话，awaitTermination()依然锁在等待终止状态，而 shutdown() 也无法得到锁去让线程池停止，这就形成了死锁
// 6.关闭线程池
if (i == 4) {
while (!threadPoolExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
System.out.println("线程池未关闭");
}
threadPoolExecutor.shutdown();
System.out.println("线程池已经关闭");
}