详解Java线程池如何统计线程空闲时间
一灯架构 人气:0背景介绍
你刚从学校毕业后,到新公司实习,试用期又被毕业,然后你又不得不出来面试,好在面试的时候碰到个美女面试官!
面试官: 小伙子,我看你简历上写的项目中用到了线程池,你知道线程池是怎样实现复用线程的?
这面试官是不是想坑我?是不是摆明了不让我通过?
难道你不应该问线程池有哪些核心参数?每个参数具体作用是什么?
往线程池中不断提交任务,线程池的处理流程是什么?
这些才是你应该问的,这些八股文我已经背熟了,你不问,瞎问什么复用线程?
幸亏我看了一灯的八股文,听我给你背一遍!
我: 线程池复用线程的逻辑很简单,就是在线程启动后,通过while死循环,不断从阻塞队列中拉取任务,从而达到了复用线程的目的。
具体源码如下:
// 线程执行入口 public void run() { runWorker(this); } // 线程运行核心方法 final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); boolean completedAbruptly = true; try { // 1. 使用while死循环,不断从阻塞队列中拉取任务 while (task != null || (task = getTask()) != null) { // 加锁,保证thread不被其他线程中断(除非线程池被中断) w.lock(); // 2. 校验线程池状态,是否需要中断当前线程 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { beforeExecute(wt, task); Throwable thrown = null; try { // 3. 执行run方法 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); } }
runWorker方法逻辑很简单,就是不断从阻塞队列中拉取任务并执行。
面试官: 小伙子,有点东西。我们都知道线程池会回收超过空闲时间的线程,那么线程池是怎么统计线程的空闲时间的?
美女面试官的问题真刁钻,让人头疼啊!这问的也太深了吧!
没看过源码的话,真不好回答。
我: 嗯...,可能是有个监控线程在后台不停的统计每个线程的空闲时间,看到线程的空闲时间超过阈值的时候,就回收掉。
面试官: 小伙子,你的想法挺不错,逻辑很严谨,你确定线程池内部是这么实现的吗?
问得我有点不自信了,没看过源码不能瞎蒙。
我还是去瞅一眼一灯写的八股文吧。
我: 这个我知道,线程池统计线程的空闲时间的实现逻辑很简单。
阻塞队列(BlockingQueue)提供了一个poll(time, unit)
方法用来拉取数据,
作用就是: 当队列为空时,会阻塞指定时间,然后返回null。
线程池就是就是利用阻塞队列的这个方法,如果在指定时间内拉取不到任务,就表示该线程的存活时间已经超过阈值了,就要被回收了。
具体源码如下:
// 从阻塞队列中拉取任务 private Runnable getTask() { boolean timedOut = false; for (; ; ) { int c = ctl.get(); int rs = runStateOf(c); // 1. 如果线程池已经停了,或者阻塞队列是空,就回收当前线程 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // 2. 再次判断是否需要回收线程 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { // 3. 在指定时间内,从阻塞队列中拉取任务 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; // 4. 如果没有拉取到任务,就标识该线程已超时,然后就被回收 timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
面试官: 小伙子,可以啊,你是懂线程池源码的。再问你个问题,如果线程池抛异常了,也没有try/catch,会发生什么?
美女面试官你这是准备打破砂锅问到底,铁了心不让我过,是吧?
我的代码风格是很严谨的,谁写的业务代码不try/catch,也没遇到过这种情况。
让我再看一下一灯总结的八股文吧。
我: 有了,线程池中的代码如果抛异常了,也没有try/catch,会从线程池中删除这个异常线程,并创建一个新线程。
不信的话,我们可以测试验证一下:
/** * @author 一灯架构 * @apiNote 线程池示例 **/ public class ThreadPoolDemo { public static void main(String[] args) { List<Integer> list = new ArrayList<>(); // 1. 创建一个单个线程的线程池 ExecutorService executorService = Executors.newSingleThreadExecutor(); // 2. 往线程池中提交3个任务 for (int i = 0; i < 3; i++) { executorService.execute(() -> { System.out.println(Thread.currentThread().getName() + " 关注公众号:一灯架构"); throw new RuntimeException("抛异常了!"); }); } // 3. 关闭线程池 executorService.shutdown(); } }
输出结果:
pool-1-thread-1 关注公众号:一灯架构
pool-1-thread-2 关注公众号:一灯架构
pool-1-thread-3 关注公众号:一灯架构
Exception in thread "pool-1-thread-1" java.lang.RuntimeException: 抛异常了!
at com.yideng.SynchronousQueueDemo.lambda$main$0(ThreadPoolDemo.java:21)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Exception in thread "pool-1-thread-2" java.lang.RuntimeException: 抛异常了!
at com.yideng.SynchronousQueueDemo.lambda$main$0(ThreadPoolDemo.java:21)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Exception in thread "pool-1-thread-3" java.lang.RuntimeException: 抛异常了!
at com.yideng.SynchronousQueueDemo.lambda$main$0(ThreadPoolDemo.java:21)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
从输出结果中可以看出,线程名称并不是同一个,而是累加的,说明原线程已经被回收,新建了个线程。
我们再看一下源码,验证一下:
// 线程抛异常后,退出逻辑 private void processWorkerExit(ThreadPoolExecutor.Worker w, boolean completedAbruptly) { if (completedAbruptly) decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { completedTaskCount += w.completedTasks; // 1. 从工作线程中删除当前线程 workers.remove(w); } finally { mainLock.unlock(); } // 2. 中断当前线程 tryTerminate(); int c = ctl.get(); 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 } // 3. 新建一个线程 addWorker(null, false); } }
如果想统一处理异常,可以自定义线程创建工厂,在工厂里面设置异常处理逻辑。
/** * @author 一灯架构 * @apiNote 线程池示例 **/ public class ThreadPoolDemo { public static void main(String[] args) { List<Integer> list = new ArrayList<>(); // 1. 创建一个单个线程的线程池 ExecutorService executorService = Executors.newSingleThreadExecutor(runnable -> { // 2. 自定义线程创建工厂,并设置异常处理逻辑 Thread thread = new Thread(runnable); thread.setUncaughtExceptionHandler((t, e) -> { System.out.println("捕获到异常:" + e.getMessage()); }); return thread; }); // 3. 往线程池中提交3个任务 for (int i = 0; i < 3; i++) { executorService.execute(() -> { System.out.println(Thread.currentThread().getName() + " 关注公众号:一灯架构"); throw new RuntimeException("抛异常了!"); }); } // 4. 关闭线程池 executorService.shutdown(); } }
输出结果:
Thread-0 关注公众号:一灯架构
捕获到异常:抛异常了!
Thread-1 关注公众号:一灯架构
捕获到异常:抛异常了!
Thread-2 关注公众号:一灯架构
捕获到异常:抛异常了!
加载全部内容