前言:
当前各位老铁们对“java线程执行完毕”大约比较看重,大家都需要学习一些“java线程执行完毕”的相关资讯。那么小编也在网络上网罗了一些对于“java线程执行完毕””的相关文章,希望小伙伴们能喜欢,姐妹们快快来了解一下吧!说在前面
在40岁老架构师 的读者交流群(50+)中,最近有小伙伴拿到了一线互联网企业如极兔、有赞、希音的面试资格,遇到一几个很重要的面试题:
还有线程池正在执行的任务和线程,如果线程池shutdown怎么办
与之类似的、其他小伙伴遇到过的问题还有:
如果还有任务没执行,线程池被关闭了,怎么办?
这里给大家做一下系统化、体系化的线程池梳理,使得大家可以充分展示一下大家雄厚的 “技术肌肉”,让面试官爱到 “不能自已、口水直流”。
也一并把这个题目以及参考答案,收入咱们的 《尼恩Java面试宝典》V60版本,供后面的小伙伴参考,提升大家的 3高 架构、设计、开发水平。
注:本文以 PDF 持续更新,最新Java 架构笔记、面试题 的PDF文件,请后台私信【笔记】获取!!!
一:首先回顾线程池线程池的5种运行状态
ThreadPoolExecutor 使用 runState (运行状态) 变量,管理线程池的生命周期,
runState 一共有以下5种取值:
(1)RUNNING:接收新的任务,并对任务队列里的任务进行处理;
(2)SHUTDOWN:不再接收新的任务,但是会对任务队列中的任务进行处理;
(3)STOP:不接收新任务,也不再对任务队列中的任务进行处理,并中断正在处理的任务;
(4)TIDYING:所有任务都已终止,线程数为0,在转向TIDYING状态的过程中,线程会执行terminated()钩子方法,钩子方法是指在本类中是空方法,而在子类中进行具体实现的方法;
(5)TERMINATED:terminated()方法执行结束后会进入这一状态,表示线程池已关闭。
与线程池关闭有关的状态,不是1个,而是有4个:
状态(2)SHUTDOWN:不再接收新的任务,但是会对任务队列中的任务进行处理;
状态(3)STOP:不接收新任务,也不再对任务队列中的任务进行处理,并中断正在处理的任务;
状态(4)TIDYING:所有任务都已终止,线程数为0,在转向TIDYING状态的过程中,线程会执行terminated()钩子方法,钩子方法是指在本类中是空方法,而在子类中进行具体实现的方法;
状态(5)TERMINATED:terminated()方法执行结束后会进入这一状态,表示线程池已彻底关闭。
从这么多的状态可以知道,线程池的关闭,不是一个简单的问题了。
二:线程池停止相关的五个方法
线程池停止相关的五个方法:
(1)shutdown方法:柔和关闭线程池;
(2)shutdownNow方法:暴力关闭线程池,无论线程池中是否有剩余任务,立刻彻底停止线程池
(3)isShutdown方法:查看线程池是否已进入停止状态了
(4)isTerminated方法:查看线程池是否已经彻底停止了
(5)awaitTermination方法:判断在等待的时间内,线程池是否彻底停止
其中终止线程池主要有2个:
(1)shutdown方法:柔和关闭线程池;
shutdown()后线程池将变成shutdown状态,此时不接收新任务,但会处理完正在运行的 和 在 workQueue 阻塞队列中等待处理的任务。
(2)shutdownNow方法:暴力关闭线程池
无论线程池中是否有剩余任务,shutdownNow()立刻彻底停止线程池。shutdownNow()后线程池将变成stop状态,此时不接收新任务,不再处理在阻塞队列中等待的任务,还会尝试中断正在处理中的工作线程。
其中对线程池关闭状态进行检查的方法,主要有3个:
(3)isShutdown方法:查看线程池是否已进入停止状态了
(4)isTerminated方法:查看线程池是否已经彻底停止了
(5)awaitTermination方法:判断在等待的时间内,线程池是否彻底停止
(1)shutdown柔和关闭线程池;
shutdown柔和关闭线程池,有两个要点:
(1)shutdown方法是关闭线程池;
(2)但是,shutdown只是初始化整个关闭过程, 执行完这个方法后,线程池不一定会立即停止;
所以,在我们调用了shutdown方法后,线程池就知道了 停止线程池的意图;而并不是我们调用shutdown方法后,整个线程池就能停的。比如,线程池在执行到一半时,线程中有正在执行的任务,队列中也可能有等待被执行的任务,线程池需要等这些任务执行完了,才能真正停止。
当然,在我们调用了shutdown方法后,如果还有新的任务过来,线程池就会拒绝。
演示案例,在《Java 高并发核心编程 卷2 加强版》随时源码中,有大量的 shutdown 使用案例。
在超级牛逼的rocketmq 源码中,也是shutdown 关闭线程池,具体如下:
说明:
(1)还是强调一下:我们执行了shutdown方法,isShutdown方法就会返回true;isShutdown方法返回true,仅仅代表线程池处于停止状态了,不代表线程池彻底停止了(因为,线程池进入停止状态后,还要等待【正在执行的任务以及队列中等待的任务】都执行完后,才能彻底终止);
(2)那么怎么看,线程池是否彻底停止了呐? 稍微晚点,要讲isTerminated()方法,可以实现这个需求;
(2)shutdownNow 粗暴关闭线程
shutdownNow方法:无论线程池中是否有剩余任务,立刻彻底停止线程池;
如何一个粗暴法呢?
(1) 正在执行任务的线程会被中断;
(2)队列中正在排队的任务,会返回;
来看一个例子:向3个线程的固定大小线程池, 提交10个任务,每个任务 500ms!
执行结果如下:
另外还有 7个 任务,没有来得及执行。
如果数据和任务都不重要,可以 shutdownNow 粗暴关闭线程,否则,这就太野蛮了。
(3)isShutdown方法:查看线程池是否已进入停止状态了;
当调用shutdown方法关闭线程后,线程不是立即关闭,仅仅是启动了关闭流程,不再接收新的任务;
问题是,如何查看线程池是否已进入停止状态呢? 难道,我们只有通过 向线程池添加任务的方式 才能看到shutdown确确实实被执行了吗?
可以通过 isShutdown方法 查看线程池是否已进入停止状态了。只要开始执行了shutdown方法,isShutdown方法就会返回true;
(4)isTerminated方法:判停, 注意是阻塞判停
threadPool.isTerminated方法:查看线程池是否已经彻底停止了
threadPool.isTerminated() 常用来判断线程池是否结束,线程池pool的状态是否为Terminated,如果是,表示线程池pool彻底终止, threadPool.isTerminated() 返回为TRU
当需要用到isTerminated()函数判断线程池中的所有线程是否执行完毕时候,不能直接使用该函数,
必须在shutdown()方法关闭线程池之后才能使用,否则isTerminated()永不为TRUE,而且线程将一直阻塞在该判断的地方,导致程序最终崩溃。
(5)awaitTermination 等待停止
awaitTermination方法:判断在等待的时间内,线程池是否彻底停止。awaitTermination第一个参数是long类型的超时时间,第二个参数可以为该时间指定单位。
awaitTermination 的功能如下:
阻塞当前线程,等已提交和已执行的任务都执行完,解除阻塞当等待超过设置的时间,检查线程池是否停止,如果停止返回 true,否则返回 false,并解除阻塞
awaitTermination 一般与shutdown()方法结合使用,下面是一个例子:
执行结果如下:
例子中,线程池的有效执行时间为20S,20S之后不管子任务有没有执行完毕,都要关闭线程池。
注意:
与shutdown()方法结合使用时,尤其要注意的是shutdown()方法必须要在awaitTermination()方法之前调用,该方法才会生效。否则会造成死锁。
关闭线程池的正确姿势
关闭线程池的正确姿势= shutdown方法 +awaitTermination方法 组合关闭。
(1)shutdown方法:柔和的关闭ExecutorService,
当此方法被调用时,pool停止接收新的任务并且等待已经提交的任务(包含提交正在执行和提交未执行)执行完成。当所有提交任务执行完毕,线程池即被关闭。
(2)awaitTermination 方法:
接收人timeout和TimeUnit两个参数,用于设定超时时间及单位。
当等待超过设定时间时,会监测ExecutorService是否已经关闭,若关闭则返回true,否则返回false。
三:线程池关闭的源码分析
接下来,分析一下线程池关闭相关的方法的源码,包括各个方法之间的逻辑关系,调用关系和产生的效果。
再次回顾:线程池的5种运行状态
ThreadPoolExecutor 使用 runState (运行状态) 变量,管理线程池的生命周期,
线程池关闭过程中,会涉及到 频繁的 runState 运行状态转化,
所以,首先需要了解线程池的各种 runState 运行状态及 各种 runState 之间的转化关系,
runState 一共有以下5种取值:
(1)RUNNING:接收新的任务,并对任务队列里的任务进行处理;
(2)SHUTDOWN:不再接收新的任务,但是会对任务队列中的任务进行处理;
(3)STOP:不接收新任务,也不再对任务队列中的任务进行处理,并中断正在处理的任务;
(4)TIDYING:所有任务都已终止,线程数为0,在转向TIDYING状态的过程中,线程会执行terminated()钩子方法,钩子方法是指在本类中是空方法,而在子类中进行具体实现的方法;
(5)TERMINATED:terminated()方法执行结束后会进入这一状态,表示线程池已关闭。
运行状态的转化条件和转化关系如下所示:
shutdown操作之后,经历三个状态:
(1)首先最重要的一点变化就是线程池状态变成了SHUTDOWN。
该状态是开始关闭线程池之后,从RUNNING改变状态经过的第一个状态,
(2)等任务队列和线程数为0之后,进入TIDYING第2个状态,
(3)等内部调用的terminated()方法执行结束后,会进入TERMINATED状态,表示线程池已关闭
shutdownNow操作之后,经历3个状态:
(1)直接进STOP,不管任务队列里边是否还有任务要处理,尝试停止所有活动的正在执行的任务,停止等待任务的处理,并排空任务列表
(2)等任务队列和线程数为0之后,进入TIDYING第2个状态,
(3)等内部调用的terminated()方法执行结束后,会进入TERMINATED状态,表示线程池已关闭
源码分析1:shutdown()柔和终止线程池
shutdown()柔和终止线程池的核心流程如下:
step1、抢占线程池的主锁
线程池的主锁是 mainLock ,是可重入锁,
当要操作workers set这个保持线程的HashSet时,需要先获取 mainLock,
另外,当要处理largestPoolSize、completedTaskCount这类统计数据时需要先获取mainLock
step 2、权限校验
java 安全管理器校验 , 判断调用者是否有权限shutdown线程池
step 3、更新线程池状态为shutdown
使用CAS操作将线程池状态设置为shutdown,
shutdown之后将不再接收新任务
step 4、中断所有空闲线程
调用 interruptIdleWorkers() 打断所有的空闲工作线程,即workerQueue.take()阻塞的线程
step 5、onShutdown(),
调用子类回调方法,基类默认为空方法
子类回调方法 可以在shutdown()时做一些处理
子类 ScheduledThreadPoolExecutor中实现了这个方法,
step 6、解锁
step 7、尝试终止线程池 tryTerminate()
public void shutdown() { final ReentrantLock mainLock = this.mainLock; // step1、抢占线程池的主锁 mainLock.lock(); try { // step 2、权限校验 java 安全管理器校验 checkShutdownAccess(); //step 3、更新线程池状态为shutdown advanceRunState(SHUTDOWN); // step 4、 打断所有的空闲工作线程,即workerQueue.take()阻塞的线程 interruptIdleWorkers(); // 调用子类回调方法,基类默认为空方法 onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { //**step 6、解锁** mainLock.unlock(); } //step 7、尝试终止线程池 tryTerminate() tryTerminate();}最重要的3个步骤是:
step 3 更新线程池状态为shutdown
step 4 中断所有空闲线程、
step 7 tryTerminated()尝试终止线程池
接下来,介绍step 4 、step 7的核心源码
step4: 中断所有空闲线程 interruptIdleWorkers()
step4 是 调用 interruptIdleWorkers() 中断所有空闲线程 完成的。有两个问题:
(1)什么是空闲线程?
(2)interruptIdleWorkers() 是怎么中断空闲线程的?
/** * 中断唤醒后,可以判断线程池状态是否变化来决定是否继续 * * onlyOne如果为true,最多interrupt一个worker * 只有当终止流程已经开始, * 但线程池还有worker线程时,tryTerminate()方法会做调用onlyOne为true的调用 * (终止流程已经开始指的是:shutdown状态 且 workQueue为空,或者 stop状态) * 在这种情况下,最多有一个worker被中断,为了传播shutdown信号,以免所有的线程都在等待 * 为保证线程池最终能终止,这个操作总是中断一个空闲worker * 而shutdown()中断所有空闲worker,来保证空闲线程及时退出 */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(); } } if (onlyOne) break; } } finally { mainLock.unlock(); //解锁 }}
interruptIdleWorkers() 首先会获取mainLock锁,因为要迭代workers 集合 ,
然后,中断在等待任务的线程(没有上锁的),在中断每个worker前,需要做两个判断:
1、线程是否已经被中断,是就什么都不做
2、worker.tryLock() 是否成功
第二个判断worker.tryLock()比较重要,因为Worker类除了实现了可执行的Runnable,也继承了AQS,
也就说,worker 本身也是一把锁.
尼恩提示,AQS的知识,非常重要,具体请阅读 《Java 高并发核心编程 卷2 加强版》。
该书对 AQS 作为浅显易懂的介绍, 被很多小伙伴称之为最为易懂的版本,pdf 是免费获取的。
worker.tryLock() 为什么要获取worker的锁呢?
Woker类在执行任务的工作线程, 都是上了worker锁的。
在 runWorker()方法中, worker 从pool 中获取task 并执行,但是执行的过程中,涉及到锁:
(1)一个worker 每次通过 getTask() 方法从 pool 获取到task 之后,在执行 task.run() 之前,都需要 worker.lock()上锁,
(2)task 运行结束后 unlock 解锁,
所以说, 只要是 正在执行任务的工作线程, 都是上了worker锁的
参考的源码如下:
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { w.lock(); // If pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted. This // requires a recheck in second case to deal with // shutdownNow race while clearing interrupt if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) 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); }}
回顾一下前面 interruptIdleWorkers 的代码,有一个核心要点:
interruptIdleWorkers 中断 work 线程之前, 需要先work.tryLock()获取worker锁,
这就 意味着正在执行task的worker线程,不能被中断。
为啥呢? worker 锁比较特殊: 核心的要点是 worker 锁是不可重入的 , 所以 不管是不是 当前线程,worker.tryLock() 都失败。
怎么证明 worker 锁是不可重入的,可以去看源码: worker 是线程池 ThreadPoolExecutor 的内部类,继承了 AbstractQueuedSynchronizer 抽象队列同步器, 核心的方法如下:
这里关键的知识点,还是AQS
所以 AQS 非常重要,具体请阅读 《Java 高并发核心编程 卷2 加强版》。
该书对 AQS 作为浅显易懂的介绍, 被很多小伙伴称之为最为易懂的版本,pdf 是免费获取的。
所以说,shutdown() 只有对能获取到worker锁的空闲线程发送中断信号, 对于忙的worker线程, 要等到拿到锁之后,才能去发中断信号。
由此可以 将worker划分为:
1、闲的worker:没有执行任务的worker,比如正在从workQueue阻塞队列中获取任务的worker,
2、忙的worker:正在task.run()执行任务的worker
线程被中断之后,如何处理
还有一点需要注意: 对于闲着的但是正在被阻塞在getTask()的worker,是可以被中断的,但是在被中断后会抛出InterruptedException,runWorker的while循环被破坏,从而 不再阻塞获取任务
worker 捕获中断异常后,将跳出 while循环,进入 processWorkerExit 方法,
runWorker 的核心代码如下:
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { w.lock(); // If pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted. This // requires a recheck in second case to deal with // shutdownNow race while clearing interrupt if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { //...... 省略n行,这里 执行拿到的任务,并处理任务异常 } finally { task = null; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { // InterruptedException 中断发生之后, 走到这里 processWorkerExit(w, completedAbruptly); }}
这里特别注意, 一旦出了那个while 循环, 这个thread 的执行,即将 结束了
换句话说,一旦worker 捕获中断异常后,worker 所绑定的thread将跳出 while循环,即将 结束了
具体请参考下面:
虽然worker 绑定的线程,即将 结束了。但是在结束之前,还要执行一下 processWorkerExit方法
processWorkerExit方法解析
来看看 processWorkerExit(Worker w, boolean completedAbruptly) 方法解析
1.参数说明:
Worker w : 工作线程包装器。
boolean completedAbruptly :默认值为true,只有调用getTask()方法,返回null,线程正常退出,会将completedAbruptly设置为false。当task.run()任务运行过程中抛出异常,线程异常退出,completedAbruptly还是默认值true。
2.执行过程:
统计执行完成的任务个数。tryTerminate() 尝试调用terminated()方法。RUNNING | SHUTDOWN 状态下,保证工作线程数量 >= corePoolSize,如果不满足,添加新线程。
private void processWorkerExit(Worker w, boolean completedAbruptly) { // 线程异常退出,修改工作线程数量。 if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 统计执行完成任务个数 completedTaskCount += w.completedTasks; // 移除当前worker workers.remove(w); } finally { mainLock.unlock(); } // 尝试调用terminated() 方法 tryTerminate(); int c = ctl.get(); //如果线程 为 RUNNING | SHUTDOWN 状态下 , 要保证最小工作线程数。 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 } // 添加新线程,在RUNNING | SHUTDOWN 状态下,不至于一个线程也没有了,要保证 剩余的任务干完 addWorker(null, false); }}
核心的工作为:
(1) 从pool 的 workers 集合移除当前worker
//Set containing all worker threads in pool. Accessed only when holding mainLock. private final HashSet<Worker> workers = new HashSet<Worker>();
(2)尝试调用 pool的 terminated() 方法
这个方法中,首先判断 pool的状态,如果为 RUNNING || (线程池已经被关闭【TIDYING | TERMINATED】) || (SHUTDOWN && 任务队列不为空),直接返回。
这个方法中,然后判断 工作线程数,如果不为0(自己不是最后一个工作线程), 随机打断一个空闲线程,直接返回。
否则,这一个线程修改线程池状态为TIDYING,修改线程状态为TERMINATED,调用terminated()方法,唤醒等待pool终止的线程 ,也就是awaitTermination() 的线程。
尼恩提示:pool的 terminated() 方法,稍微晚点介绍。
(3) 保证最小工作线程数
上面的代码中,使用runStateLessThan(c, STOP) 判断线程的状态 是否比 STOP 小,那么比STOP 小的是谁呢?
(1)RUNNING状态
(2)SHUTDOWN 状态
ThreadPoolExecutor用一个AtomicInteger字段保存了2个状态
workerCount (有效线程数) (占用29位)runState (线程池运行状态) (占用高3位)
//标记线程数和状态的混合值private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//线程位数private static final int COUNT_BITS = Integer.SIZE - 3;//线程最大个数(低29位)00011111111111111111111111111111private static final int COUNT_MASK = (1 << COUNT_BITS) - 1; //(高3位):11100000000000000000000000000000private static final int RUNNING = -1 << COUNT_BITS;//(高3位):00000000000000000000000000000000private static final int SHUTDOWN = 0 << COUNT_BITS;//(高3位):00100000000000000000000000000000private static final int STOP = 1 << COUNT_BITS;//(高3位):01000000000000000000000000000000private static final int TIDYING = 2 << COUNT_BITS;//(高3位):01100000000000000000000000000000private static final int TERMINATED = 3 << COUNT_BITS; //获取线程池运行状态private static int runStateOf(int c) { return c & ~COUNT_MASK; }//获取线程个数private static int workerCountOf(int c) { return c & COUNT_MASK; }//计算ctl新值private static int ctlOf(int rs, int wc) { return rs | wc; }
从上面的源码可以看出 ,比STOP 小的是RUNNING | SHUTDOWN
processWorkerExit方法需要保证:如果pool在 RUNNING | SHUTDOWN 状态下,不能一个线程也没有了,要保证 workQueue 剩余的任务干完
所以,在RUNNING | SHUTDOWN 状态下, 如果有必要,还要添加新线程,
step7: 尝试终止线程池 tryTerminate()
shutdown()的最后也调用了tryTerminated()方法,下面看看这个方法的逻辑:
/** * 在以下情况将线程池变为TERMINATED终止状态 * shutdown 且 正在运行的worker 和 workQueue队列 都empty * stop 且 没有正在运行的worker * * 这个方法必须在任何可能导致线程池终止的情况下被调用,如: * 减少worker数量 * shutdown时从queue中移除任务 * * 这个方法不是私有的,所以允许子类ScheduledThreadPoolExecutor调用 */final void tryTerminate() { //这个for循环主要是和进入关闭线程池操作的CAS判断结合使用的 for (;;) { int c = ctl.get(); /** * 线程池是否需要终止 * 如果以下3中情况任一为true,return,不进行终止 * 1、还在运行状态 * 2、状态是TIDYING、或 TERMINATED,已经终止过了 * 3、SHUTDOWN 且 workQueue不为空 */ if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return; /** * 只有shutdown状态 且 workQueue为空,或者 stop状态能执行到这一步 * 如果此时线程池还有线程(正在运行任务,正在等待任务) * 中断唤醒一个正在等任务的空闲worker * 唤醒后再次判断线程池状态,会return null,进入processWorkerExit()流程 */ if (workerCountOf(c) != 0) { // Eligible to terminate 资格终止 interruptIdleWorkers(ONLY_ONE); //中断workers集合中的空闲任务,参数为true,只中断一个 return; } /** * 如果状态是SHUTDOWN,workQueue也为空了,正在运行的worker也没有了,开始terminated */ final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //CAS:将线程池的ctl变成TIDYING(所有的任务被终止,workCount为0, // 为此状态时将会调用terminated()方法),期间ctl有变化就会失败,会再次for循环 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { try { terminated(); //需子类实现 } finally { ctl.set(ctlOf(TERMINATED, 0)); //将线程池的ctl变成TERMINATED termination.signalAll(); //唤醒调用了 等待线程池终止的线程 awaitTermination() } return; } } finally { mainLock.unlock(); } // else retry on failed CAS // 如果上面的CAS判断false,再次循环 }}
tryTerminate() 执行流程:
1、判断线程池是否需要进入终止流程(只有当shutdown状态+workQueue.isEmpty 或 stop状态,才需要)
2、判断线程池中是否还有线程,有则interruptIdleWorkers(ONLY_ONE)尝试中断一个空闲线程
正是这个逻辑可以再次发出中断信号,中断阻塞在获取任务的线程
3、如果状态是SHUTDOWN,workQueue也为空了,正在运行的worker也没有了,开始terminated
会先上锁,将线程池置为tidying状态,之后调用需子类实现的 terminated(),最后线程池置为terminated状态,并唤醒所有等待线程池终止这个Condition的线程
源码分析2:shutdownNow() 粗暴终止线程池的核心流程
/** * 尝试停止所有活动的正在执行的任务,停止等待任务的处理,并返回正在等待被执行的任务列表 * 这个任务列表是从任务队列中排出(删除)的 * * 这个方法不用等到正在执行的任务结束,要等待线程池终止可使用awaitTermination() * * 除了尽力尝试停止运行中的任务,没有任何保证 * 取消任务是通过Thread.interrupt()实现的,所以任何响应中断失败的任务可能永远不会结束 */public List <Runnable> shutdownNow() { List <Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); //上锁 try { //判断调用者是否有权限shutdown线程池 checkShutdownAccess(); //CAS+循环设置线程池状态为stop advanceRunState(STOP); //中断所有线程,包括正在运行任务的 interruptWorkers(); tasks = drainQueue(); //将workQueue中的元素放入一个List并返回 } finally { mainLock.unlock(); //解锁 } //尝试终止线程池 tryTerminate(); return tasks; //返回workQueue中未执行的任务}
shutdownNow() 和 shutdown()的大体流程相似,差别是:
1、将线程池更新为stop状态
2、调用interruptWorkers()中断所有线程,包括正在运行的线程
3、将workQueue中待处理的任务移到一个List中,并在方法最后返回,说明shutdownNow()后不会再处理workQueue中的任务
interruptWorkers()
private void interruptWorkers() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) w.interruptIfStarted(); } finally { mainLock.unlock(); }}
interruptWorkers() 很简单,循环对所有worker调用 interruptIfStarted(),其中会判断worker的AQS state是否大于0,即worker是否已经开始运作,再调用Thread.interrupt()
注意:
对于运行中的线程调用Thread.interrupt()并不能保证线程被终止,为啥呢?
task.run()内部执行的是业务代码,如果业务代码里边捕获了InterruptException,没有上抛,导致这里的结束机制失效。
改怎么办呢?其实也无所谓。
当runWorker 执行下一次或者任务之后,里边会进行 线程池状态的双重检查,如果线程池的状态变了,变为结束,那么 工作线程 也会被 中断了。
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { w.lock(); // If pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted. This // requires a recheck (线程池状态的双重检查) in second case to deal with // shutdownNow race while clearing interrupt if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) 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); }}源码分析3:awaitTermination() 等待线程池终止的核心流程
这个方法,也比较重要,咱们顺便看看源码吧。
awaitTermination() 源码如下
// 参数: timeout:超时时间 unit:timeout超时时间的单位//返回: true:线程池终止 , false:超过timeout指定时间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(); }}
在发出一个shutdown请求后,在以下3种情况发生之前,awaitTermination()都会被阻塞
1、所有任务完成执行
2、到达超时时间
3、当前线程被中断
这里用到一个 锁条件 termination:
/*** Wait condition to support awaitTermination*/private final Condition termination = mainLock.newCondition();
awaitTermination() 循环的判断线程池是否terminated终止 或 是否已经超过超时时间,然后通过termination这个Condition阻塞等待一段时间
termination 阻塞等待过程中发生以下具体情况会解除阻塞(对上面3种情况的解释):
1、如果发生了 termination.signalAll()(内部实现是 LockSupport.unpark())会唤醒阻塞等待,且由于ThreadPoolExecutor只有在 tryTerminated()尝试终止线程池成功,将线程池更新为terminated状态后才会signalAll(),故awaitTermination()再次判断状态会return true退出
2、如果达到了超时时间 termination.awaitNanos() 也会返回,此时nano==0,再次循环判断return false,等待线程池终止失败
3、如果当前线程被 Thread.interrupt(),termination.awaitNanos()会上抛InterruptException,awaitTermination()继续上抛给调用线程,会以异常的形式解除阻塞
故终止线程池并需要知道其是否终止,可以用如下方式:
executorService.shutdown();try{ while(!executorService.awaitTermination(500, TimeUnit.MILLISECONDS)) { LOGGER.debug("Waiting for terminate"); } } catch (InterruptedException e) { //中断处理}
标签: #java线程执行完毕 #java线程执行一半消失了怎么解决 #线程池执行任务失败怎么重新提交任务 #线程池执行任务失败怎么重新提交任务信息 #线程池执行任务失败怎么重新提交任务信息错误