线程数突增!领导说再这么写就gc掉我

语言: CN / TW / HK

线程数突增!领导说再这么写就gc掉我

前言

大家好,我是魔性的茶叶,今天给大家分享一个线上问题引出的一次思考,过程比较长,但是挺有意思。

今天上班把需求写完,出于学习(摸鱼)的心理上skywalking看看,突然发现我们的一个应用,应用内线程数超过900条,接近1000条,但是cpu并没有高涨,内存也不算高峰。但是敏锐的我还是立刻意识到这个应用有不妥,因为线程数太多了,不符合我们一个正常健康的应用数量。熟练的打出cpu dump观察,首先看线程组名的概览。

image-20230112200957387

从线程分组看,pool名开头线程占616条,而且waiting状态也是616条,这个点就非常可疑了,我断定就是这个pool开头线程池导致的问题。我们先排查为何这个线程池中会有600+的线程处于waiting状态并且无法释放,记接下来我们找几条线程的堆栈观察具体堆栈:

image-20230112201456234

这个堆栈看上去很合理,线程在线程池中不断的循环获取任务,因为获取不到任务所以进入了waiting状态,等待着有任务后被唤醒。

看上去不只一个线程池,并且这些线程池的名字居然是一样的,我大胆的猜测一下,是不断的创建同样的线程池,但是线程池无法被回收导致的线程数,所以接下来我们要分析两个问题,首先这个线程池在代码里是哪个线程池,第二这个线程池是怎么被创建的?为啥释放不了?

我在idea搜索new ThreadPoolExecutor()得到的结果是这样的:

image-20230112202915219

于是我陷入懵逼的状态,难道还有其他骚操作?

正在这时,一位不知名的郑网友发来一张截图:

image-20230112203527173

好家伙!竟然是用new FixedTreadPool()整出来的。难怪我完全搜不到,因为用的new FixedTreadPool(),所以线程池中的线程名是默认的pool(又多了一个不使用Executors来创建线程池的理由)。

然后我迫不及die的打开代码,试图找到罪魁祸首,结果发现作者居然是我自己。这是另一个惊喜,惊吓的惊。

冷静下来后我梳理一遍代码,这个接口是我两年前写的,主要是功能是统计用户的钱包每个月的流水,因为担心统计比较慢,所以使用了线程池,做了批量的处理,没想到居然导致了线程数过高,虽然没有导致事故,但是确实是潜在的隐患,现在没出事不代表以后不会出事。

去掉多余业务逻辑,我简单的还原一个代码给大家看,还原现场:

private static void threadDontGcDemo(){       ExecutorService executorService = Executors.newFixedThreadPool(10);       executorService.submit(() -> {            System.out.println("111");        });    }

那么为啥线程池里面的线程和线程池都没释放呢

难道是因为没有调用shutdown?我大概能理解我两年前当时为啥不调用shutdown,是因为当初我觉得接口跑完,方法走到结束,理论上栈帧出栈,局部变量应该都销毁了,按理说executorService这个变量应该直接GG了,那么按理说我是不用调用shutdown方法的。

我简单的跑了个demo,循环的去new线程池,不调用shutdown方法,看看线程池能不能被回收

image-20230113142322106

打开java visual vm查看实时线程:

image-20230113142304644

可以看到线程数和线程池都一直在增加,但是一直没有被回收,确实符合发生的问题状况,那么假如我在方法结束前调用shutdown方法呢,会不会回收线程池和线程呢?

简单写个demo结合jvisualvm验证下:

image-20230113142902514

image-20230113142915722

结果是线程和线程池都被回收了。也就是说,执行了shutdown的线程池最后会回收线程池和线程对象

我们知道,一个对象能不能回收,是看它到gc root之间有没有可达路径,线程池不能回收说明到达线程池的gc root还是有可达路径的。这里讲个冷知识,这里的线程池的gc root是线程,具体的gc路径是thread->workers->线程池。线程对象是线程池的gc root,假如线程对象能被gc,那么线程池对象肯定也能被gc掉(因为线程池对象已经没有到gc root的可达路径了)。

那么现在问题就转为线程对象是在什么时候gc

郑网友给了一个粗浅但是合理的解释,线程对象肯定不是在运行中的时候被回收的,因为jvm肯定不可能去回收一条在运行中的线程,至少runnalbe状态的线程jvm不可能去回收。

在stackoverflow上我找到了更准确的答案:https://stackoverflow.com/questions/2423284/java-thread-garbage-collected-or-not

image-20230113152802164

A running thread is considered a so called garbage collection root and is one of those things keeping stuff from being garbage collected。

这句话的意思是,一条正在运行的线程是gc root,注意,是正在运行,这个正在运行我先透露下,即使是waiting状态,也算正在运行。这个回答的整体的意思是,运行的线程是gc root,但是非运行的线程不是gc root(可以被回收)。

现在比较清楚了,线程池和线程被回收的关键就在于线程能不能被回收,那么回到原来的起点,为何调用线程池的shutdown方法能够导致线程和线程池被回收呢?难道是shutdown方法把线程变成了非运行状态吗

talk is cheap,show me the code

我们直接看看线程池的shutdown方法的源码

public void shutdown() {        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            checkShutdownAccess();            advanceRunState(SHUTDOWN);            interruptIdleWorkers();            onShutdown(); // hook for ScheduledThreadPoolExecutor       } finally {            mainLock.unlock();       }        tryTerminate(); } ​ private void interruptIdleWorkers() {        interruptIdleWorkers(false); } ​ 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方法入手,这方法看上去最可疑,看到interruptIdleWorkers方法,这个方法里面主要就做了一件事,遍历当前线程池中的线程,并且调用线程的interrupt()方法,通知线程中断,也就是说shutdown方法只是去遍历所有线程池中的线程,然后通知线程中断。所以我们需要了解线程池里的线程是怎么处理中断的通知的。

我们点开worker对象,这个worker对象是线程池中实际运行的线程,所以我们直接看worker的run方法,中断通知肯定是在里面被处理了

//WOrker的run方法里面直接调用的是这个方法 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);       } } ​ ​

这个runwoker属于是线程池的核心方法了,相当的有意思,线程池能不断运作的原理就是这里,我们一点点看。

首先最外层用一个while循环套住,然后不断的调用gettask()方法不断从队列中取任务,假如拿不到任务或者任务执行发生异常(抛出异常了)那就属于异常情况,直接将completedAbruptly 设置为true,并且进入异常的processWorkerExit流程。

我们看看gettask()方法,了解下啥时候可能会抛出异常:

private Runnable getTask() {        boolean timedOut = false; // Did the last poll() time out? ​        for (;;) {            int c = ctl.get();            int rs = runStateOf(c); ​            // Check if queue empty only if necessary.            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {                decrementWorkerCount();                return null;           } ​            int wc = workerCountOf(c); ​            // Are workers subject to culling?            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; ​            if ((wc > maximumPoolSize || (timed && timedOut))                && (wc > 1 || workQueue.isEmpty())) {                if (compareAndDecrementWorkerCount(c))                    return null;                continue;           } ​            try {                Runnable r = timed ?                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                    workQueue.take();                if (r != null)                    return r;                timedOut = true;           } catch (InterruptedException retry) {                timedOut = false;           }       }   }

这样很清楚了,抛去前面的大部分代码不看,这句代码解释了gettask的作用:

Runnable r = timed ?   workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :   workQueue.take()

gettask就是从工作队列中取任务,但是前面还有个timed,这个timed的语义是这样的:如果allowCoreThreadTimeOut参数为true(一般为false)或者当前工作线程数超过核心线程数,那么使用队列的poll方法取任务,反之使用take方法。这两个方法不是重点,重点是poll方法和take方法都会让当前线程进入time_waiting或者waiting状态。而当线程处于在等待状态的时候,我们调用线程的interrupt方法,毫无疑问会使线程当场抛出异常

也就是说线程池的shutdownnow方法调用interruptIdleWorkers去对线程对象interrupt是为了让处于waiting或者是time_waiting的线程抛出异常

那么线程池是在哪里处理这个异常的呢?我们看runwoker中的调用的processWorkerExit方法,说实话这个方法看着就像处理抛出异常的方法:

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;            workers.remove(w);       } finally {            mainLock.unlock();       } ​        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           }            addWorker(null, false);       } }

我们可以看到,在这个方法里有一个很明显的 workers.remove(w)方法,也就是在这里,这个w的变量,被移出了workers这个集合,导致worker对象不能到达gc root,于是workder对象顺理成章的变成了一个垃圾对象,被回收掉了。然后等到worker中所有的worker都被移出works后,并且当前请求线程也完成后,线程池对象也成为了一个孤儿对象,没办法到达gc root,于是线程池对象也被gc掉了。

写了挺长的篇幅,我小结一下:

  1. 线程池调用shutdownnow方法是为了调用worker对象的interrupt方法,来打断那些沉睡中的线程(waiting或者time_waiting状态),使其抛出异常
  2. 线程池会把抛出异常的worker对象从workers集合中移除引用,此时被移除的worker对象因为没有到达gc root的路径已经可以被gc掉了
  3. 等到workers对象空了,并且当前tomcat线程也结束,此时线程池对象也可以被gc掉,整个线程池对象成功释放

最后总结:

如果只是在局部方法中使用线程池,线程池对象不是bean的情况时,记得要合理的使用shutdown或者shutdownnow方法来释放线程和线程池对象,如果不使用,会造成线程池和线程对象的堆积。