| 有赞一面：还有任务没执行，线程池被关闭怎么办？ | 您所在的位置：网站首页 › etc充值功能已关闭怎么办 › 有赞一面：还有任务没执行，线程池被关闭怎么办？ |
有赞一面：还有任务没执行，线程池被关闭怎么办？
|
说在前面
在40岁老架构师 尼恩的读者交流群(50+)中，最近有小伙伴拿到了一线互联网企业如极兔、有赞、希音的面试资格，遇到一几个很重要的面试题： 还有线程池正在执行的任务和线程，如果线程池shutdown怎么办？ 与之类似的、其他小伙伴遇到过的问题还有： 如果还有任务没执行，线程池被关闭了，怎么办？ 这里尼恩给大家做一下系统化、体系化的线程池梳理，使得大家可以充分展示一下大家雄厚的 “技术肌肉”，让面试官爱到 “不能自已、口水直流”。 也一并把这个题目以及参考答案，收入咱们的 《尼恩Java面试宝典》V60版本，供后面的小伙伴参考，提升大家的 3高 架构、设计、开发水平。 注：本文以 PDF 持续更新，最新尼恩 架构笔记、面试题 的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() 返回为TRUE
当需要用到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 workers = new HashSet();（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位)00011111111111111111111111111111 private static final int COUNT_MASK = (1 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 shutdownNow() { List 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 while(!executorService.awaitTermination(500, TimeUnit.MILLISECONDS)) { LOGGER.debug("Waiting for terminate"); } } catch (InterruptedException e) { //中断处理 } 参考文献：清华大学出版社 《尼恩 Java 高并发核心编程 卷2 加强版》 4000页《尼恩Java面试宝典》中 专题29 多线程 面试专题 推荐阅读： 尼恩N篇硬核架构 文章 ，帮你实现 架构自由：《吃透8图1模板，人人可以做架构》 《10Wqps评论中台，如何架构？B站是这么做的！！！》 《阿里二面：千万级、亿级数据，如何性能优化？ 教科书级 答案来了》 《峰值21WQps、亿级DAU，小游戏《羊了个羊》是怎么架构的？》 《100亿级订单怎么调度，来一个大厂的极品方案》 《2个大厂 100亿级 超大流量 红包 架构方案》 … 更多架构文章，正在添加中 实现你的响应式 自由：《响应式圣经：10W字，实现Spring响应式编程自由》 这是老版本 《Flux、Mono、Reactor 实战（史上最全）》 实现你的 spring cloud 自由：《sentinel （史上最全）》 《Nacos (史上最全)》 《SpringCloud+Dubbo3 = 王炸 ！》 《Springcloud gateway 底层原理、核心实战 (史上最全)》 《nacos高可用（图解+秒懂+史上最全）》 《分库分表 Sharding-JDBC 底层原理、核心实战（史上最全）》 《一文搞定：SpringBoot、SLF4j、Log4j、Logback、Netty之间混乱关系（史上最全）》 实现你的 linux 自由：《Linux命令大全：2W多字，一次实现Linux自由》 实现你的 网络 自由：《TCP协议详解 (史上最全)》 《网络三张表：ARP表, MAC表, 路由表，实现你的网络自由！！》 实现你的 分布式锁 自由：《Redis分布式锁（图解 - 秒懂 - 史上最全）》 《Zookeeper 分布式锁 - 图解 - 秒懂》 实现你的 王者组件 自由：《队列之王： Disruptor 原理、架构、源码 一文穿透》 《缓存之王：Caffeine 源码、架构、原理（史上最全，10W字 超级长文）》 《缓存之王：Caffeine 的使用（史上最全）》 《Java Agent 探针、字节码增强 ByteBuddy（史上最全）》 实现你的 面试题 自由：《阿里一面：你做过哪些代码优化？来一个人人可以用的极品案例》 《网易二面：CPU狂飙900%，该怎么处理？》 《场景题：假设10W人突访，你的系统如何做到不 雪崩？》 《Nginx面试题（史上最全 + 持续更新）》 《K8S面试题（史上最全 + 持续更新）》 《操作系统面试题（史上最全、持续更新）》 《Docker面试题（史上最全 + 持续更新）》 《clickhouse 超底层原理 + 高可用实操 （史上最全）》 《环形队列、 条带环形队列 Striped-RingBuffer （史上最全）》 《单例模式（史上最全）》 《红黑树（ 图解 + 秒懂 + 史上最全）》 《分布式事务 （秒懂）》 《Docker原理（图解+秒懂+史上最全）》 《Zookeeper Curator 事件监听 - 10分钟看懂》 《Netty 粘包 拆包 | 史上最全解读》 《Netty 100万级高并发服务器配置》 《Springcloud 高并发 配置 （一文全懂）》 |
【本文地址】