龙空技术网

有赞一面:还有任务没执行,线程池被关闭怎么办?

Java码农之路 689

前言:

当前各位老铁们对“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线程执行一半消失了怎么解决 #线程池执行任务失败怎么重新提交任务 #线程池执行任务失败怎么重新提交任务信息 #线程池执行任务失败怎么重新提交任务信息错误