一、说明
ThreadPoolExecutor
二、理解
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();
}
}
复制代码
线程池的生命周期
// 初始运行状态为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()
调用之后,停止接收新任务,但是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("线程池已经关闭");
}
复制代码
评论