亲宝软件园·资讯

展开

详解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 关注公众号:一灯架构
捕获到异常:抛异常了!

加载全部内容

相关教程
猜你喜欢
用户评论