线程池也会导致OOM的原因

墨蓝 2023-09-25 17:34 150阅读 0赞

### 1. 前言 ###

我这边从一个问题引出这次的话题，我们可能会在开中碰到一种OOM问题，java.lang.OutOfMemoryError: pthread\_create (1040KB stack) failed: Try again

相信很多人碰到过这个错误，很容易从网上搜索到出现这个问题的原因是线程过多，那线程过多为什么会导致OOM？线程什么情况下会释放资源？你又能如何做到让它不释放资源？

有的人可能会想到，那既然创建线程过多会导致OOM，那我用线程池不就行了。但是有没有想过，线程池，也可能会造成OOM。其实这里有个很经典的场景，你使用OkHttp的时候不注意，每次请求都创建OkHttpClient，导致线程池过多出现OOM

### 2. 简单了解线程池 ###

如何去了解线程池，看源码，直接去看是很难看得懂的，要先了解线程池的原理，对它的设计思想有个大概的掌握之后，再去看源码，就会轻松很多，当然这里只了解基础的原理还不够，还需要有一些多线程相关的基础知识。

本篇文章只从部分源码的角度去分析，线程池如何导致OOM的，而不会全部去看所有线程池的源码细节，因为太多了

首先，要了解线程池，首先需要从它的参数入手：

*  corePoolSize：核心线程数量
 *  maximumPoolSize：最大线程数量
 *  keepAliveTime，unit：非核心线程的存活时间和单位
 *  workQueue：阻塞队列
 *  ThreadFactory：线程工厂
 *  RejectedExecutionHandler：饱和策略

然后你从网上任何一个地方搜都能知道它大致的工作流程是，当一个任务开始执行时，先判断当前线程池数量是否达到核心线程数，没达到则创建一个核心线程来执行任务，如果超过，放到阻塞队列中等待，如果阻塞队列满了，未达到最大线程数，创建一条非核心线程执行任务，如果达到最大线程数，执行饱和策略。在这个过程中，核心线程不会回收，非核心线程会根据keepAliveTime和unit进行回收。

\*\*这里可以多提一嘴，这个过程用了工厂模式ThreadFactory和策略模式RejectedExecutionHandler，关于策略模式可以看我这篇文章 \*\* juejin.cn/post/719502…

其实从这里就可以看出为什么线程池也会导致OOM了：核心线程不会回收，非核心线程使用完之后会根据keepAliveTime和unit进行回收 ，那核心线程就会一直存活（我这不考虑shutdown()和shutdownNow()这些情况），一直存活就会占用内存，那你如果创建很多线程池，就会OOM。

所以我这篇文章要分析：核心线程不会释放资源的过程，它内部怎么做到的。 只从这部分的源码去进行分析，不会全部都详细讲。

先别急，为了照顾一些基础不太好的朋友，涉及一些基础知识感觉还是要多讲一下。上面提到的线程回收和shutdown方法这些是什么意思？线程执行完它内部的代码后会主动释放资源吗？

我们都知道开发中有个概念叫生命周期，当然线程池和线程也有生命周期（这很重要），在开发中，我们称之为lifecycle。

生命周期当然是设计这个东西的开发者所定义的，我们先看线程池的生命周期，在ThreadPoolExecutor的注释中有写：

*
    * The runState provides the main lifecycle control, taking on values:
    *
    *   RUNNING:  Accept new tasks and process queued tasks
    *   SHUTDOWN: Don't accept new tasks, but process queued tasks
    *   STOP:     Don't accept new tasks, don't process queued tasks,
    *             and interrupt in-progress tasks
    *   TIDYING:  All tasks have terminated, workerCount is zero,
    *             the thread transitioning to state TIDYING
    *             will run the terminated() hook method
    *   TERMINATED: terminated() has completed
    *
    复制代码

看得出它的生命周期有RUNNING，SHUTDOWN，STOP，TIDYING和TERMINATED。而shutdown()和shutdownNow()方法会改变生命周期，这里不是对线程池做全面解析，所以先有个大概了解就行，可以暂时理解成这篇文章的所有分析都是针对RUNNING状态下的。

看完线程池的，再看看线程的生命周期。线程的生命周期有：

*  NEW：创建，简单来说就是new出来没start
 *  RUNNABLE：运行，简单来说就是start后执行run方法
 *  TERMINATED：中止，简单来说就是执行完run方法或者进行中断操作之后会变成这个状态
 *  BLOCKED：阻塞，就是加锁之后竞争锁会进入到这个状态
 *  WAITING、TIMED\_WAITING：休眠，比如sleep方法

这个很重要，需要了解，你要学会线程这块相关的知识点的话，这些生命周期要深刻理解 。比如BLOCKED和WAITING有什么不同？然后学这块又会涉及到锁那一块的知识。以后有时间可以单独写几篇这类的文章，这里先大概有个概念，只需要能先看懂后面的源码就行。

从生命周期的概念你就能知道线程执行完它内部的代码后会主动释放资源，因为它run执行完之后生命周期会到TERMINATED，那这又涉及到了一个知识点，为什么主线程（ActivityThread），执行完run的代码后不会生命周期变成TERMINATED，这又涉及到Looper，就得了解Handler机制，可以看我这篇文章 juejin.cn/post/715882…

扯远了，现在进入正题，先想想，如果是你，你怎么做让核心线程执行完run之后不释放资源，很明显，只要让它不执行到TERMINATED生命周期就行，如何让它不变成TERMINATED状态，只需要让它进入BLOCKED或者WAITING状态就行。所以我的想法是这样的，当这个核心线程执行完这个任务之后，我让它WAITING，等到有新的任务进来的时候我再唤醒它进入RUNNABLE状态。 这是我从理论这个角度去分析的做法，那看看实际ThreadPoolExecutor是怎么做的

### 3. 线程池部分源码分析 ###

前面说了，不会全部都讲，这里涉及到文章相关内容的流程就是核心线程的任务执行过程，所以这里主要分析核心线程。

当我们使用线程池执行一个任务时，会调用ThreadPoolExecutor的execute方法

public void execute(Runnable command) {
        ......
    
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        
        // 我们只看核心线程的流程，所以后面的代码不用管
        ......
    }
    复制代码

这个ctl是一个状态相关的代码，可以先不用管，我后面会简单统一做个解释，这里不去管它会比较容易理解，我们现在主要是为了看核心线程的流程。从这里可以看出，当前线程的数量小于核心线程的话执行addWorker方法

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
    
            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
    
            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
    
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());
    
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
    复制代码

这个addWorker分为上下两部分，我们分别来做解析

这里主要是做了状态判断的一些操作，我说过状态相关的我们可以先不管，但是这里的写法我觉得要单独讲一下为什么会这么写。不然它内部很多代码是这样的，我怕劝退很多人。

首先retry: ...... break retry; 这个语法糖，平常我们开发很少用到，可以去了解一下，这里就是为了跳出循环。 其次，这里的compareAndIncrementWorkerCount内部的代码是AtomicInteger ctl.compareAndSet(expect, expect + 1) ，Atomic的compareAndSet操作搭配死循环，这叫自旋，所以说要看懂这个需要一定的java多线程相关的基础。自旋的目的是为了什么？这就又涉及到了锁的分类中有乐观锁，有悲观锁。不清楚的可以去学一下这些知识，你就知道为什么它要这么做了，这里就不一一解释。包括你看它的源码，能看到，它会很多地方用自旋，很多地方用ReentrantLock，但它就是不用synchronized ，这些都是多线程这块基础的知识，这里不多说了。

看看下半部分

private boolean addWorker(Runnable firstTask, boolean core) {
        
        // 上半部分
        ......
        
        
    
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                ......
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            ......
        }
        return workerStarted;
    }
    复制代码

看到它先创建一个Worker对象，再调用Worker对象内部的线程的start方法，我们看看Worker

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
    
        private static final long serialVersionUID = 6138294804551838833L;
    
        final Thread thread;
        Runnable firstTask;
    
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
    
        public void run() {
            runWorker(this);
        }
        
        // 其它方法
       ......
    }
    复制代码

看到它内部主要有两个对象firstTask就是任务，thread就是执行这个任务的线程，而这个线程是通过getThreadFactory().newThread(this)创建出来的，这个就是我们创建ThreadPoolExecutor时传的“线程工厂”  
外部调t.start();之后就会执行这里的run方法，因为newThread传了this进去，你可以先简单理解调这个线程start会执行到这个run，然后run中调用runWorker(this);

注意，你想想runWorker(this)方法，包括之后的流程，都是执行在哪个线程中？都是执行在子线程中，因为这个run方法中的代码，都是执行在这个线程中。你一定要理解这一步，不然你自己看源码会可能看懵。 因为有些人长期不接触多线程环境的情况下，你会习惯单线程的思维去看问题，那就很容易出现理解上的错误。

我们继续看看runWorker，时刻提醒你自己，之后的流程都是在子线程中进行，这条子线程的生命周期变为RUNNABLE

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) {s
                w.lock();
                
                // 中断相关的操作
                ......
                
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                        ......
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    ......
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }
    复制代码

先讲讲这里的一个开发技巧，task.run()就是执行任务，它前面的beforeExecute和afterExecute就是模板方法设计模式，方便扩展用。  
执行完任务后，最后执行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();
    
        ......
    }
    复制代码

workers.remove(w)后执行tryTerminate方法尝试将线程池的生命周期变为TERMINATED

final void tryTerminate() {
        for (;;) {
            int c = ctl.get();
            if (isRunning(c) ||
                runStateAtLeast(c, TIDYING) ||
                (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;
            if (workerCountOf(c) != 0) { // Eligible to terminate
                interruptIdleWorkers(ONLY_ONE);
                return;
            }
    
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                        terminated();
                    } finally {
                        ctl.set(ctlOf(TERMINATED, 0));
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
            // else retry on failed CAS
        }
    }
    复制代码

先不用管状态的变化，一般一眼都能看得出这里是结束的操作了，我们追踪的核心线程正常在RUNNING状态下是不会执行到这里的。 那我们期望的没任务情况下让线程休眠的操作在哪里？  
看回runWorker方法

final void runWorker(Worker w) {
        ......
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {s
                ......
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }
    复制代码

看到它的while中有个getTask()方法，认真看runWorker方法其实能看出，核心线程执行完一个任务之后会getTask()拿下一个任务去执行，这就是当核心线程满的时候任务会放到阻塞队列中，核心线程执行完任务之后会从阻塞队列中拿下一个任务执行。 getTask()从抽象上来看，就是从队列中拿任务。

private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?
    
        for (;;) {
            ......
    
            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }
    复制代码

先把timed当成正常情况下为false，然后会执行workQueue.take()，这个workQueue是阻塞队列BlockingQueue， 注意，这里又需要有点基础了。正常有点基础的人看到这里，已经知道这里就是当没有任务会让核心线程休眠的操作，看不懂的，可以先了解下什么是AQS，可以看看我这篇文章 juejin.cn/post/716801…

如果你说你懒得看，行吧，我随便拿个ArrayBlockingQueue给你举例

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }
    复制代码

notEmpty是Condition，这里调用了Condition的await()方法，然后想想执行这步操作的是在哪条线程上？线程进入WAITING状态了吧，不会进入TERMINATED了吧。

然后当有任务添加之后会唤醒它，它继续在循环中去执行任务。

这就验证了我们的猜想，通过让核心线程进入WAITING状态以此来达到执行完run方法中的任务也不会主动TERMINATED而释放线程。所以核心线程一直占用资源，这里说的资源指的是空间，而cpu的时间片是会让出的。

### 4. 部分线程池的操作解读 ###

为什么线程池也会导致OOM，上面已经通过源码告诉你，核心线程不会释放内存空间，导致线程池多的情况下也会导致OOM。这里为了方便新手阅读ThreadPoolExecutor相关的代码，还是觉得写一些它内部的设计思想，不然没点基础的话确实很难看懂。

首先就是状态，上面源码中都有关线程池的生命中周期状态（ctl字段），可以看看它怎么设计的

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private static final int COUNT_BITS = Integer.SIZE - 3;  // Integer.SIZE是32
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
    // runState is stored in the high-order bits
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;
    private static final int TIDYING    =  2 << COUNT_BITS;
    private static final int TERMINATED =  3 << COUNT_BITS;
    复制代码

它这里用了两个设计思想，第一个就是用位来表示状态，关于这类型的设计，可以看我这2篇文章 juejin.cn/post/715547… 和 juejin.cn/post/720550…

另外一个设计思想是：用一个变量的高位置表示状态，低位表示数量。 这里就是用高3位来表示生命周期，剩下的低位表示线程的数量。和这个类似的操作有view中的MeasureSpec，也是一个变量表示两个状态。

然后关于设计模式，可以看到它这里最经典的就是用了策略模式，如果你看饱和策略那块的源码，可以好好看看它是怎么设计的。其它的还有工厂、模板之类的，这些也不难，就是策略还是建议学下它怎么去设计的。

然后多线程相关的基础，这个还是比较重要的，这块的基础不好，看ThreadPoolExecutor的源码会相对吃力。比如我上面提过的，线程的生命周期，锁相关的知识，还有AQS等等。如果你熟悉这些，再看这个源码就会轻松很多。

对于总体的设计，你第一看会觉得它的源码很绕，为什么会这样？因为有中断操作+自旋锁+状态的设计 ，它的这种设计就基本可以说是优化代码到极致，比如说状态的设计，就比普通的能省内存，能更方便通过CAS操作。用自旋就是乐观锁，能节省资源等。有中断操作，能让整个系统更灵活。相对的缺点就是不安全，什么意思呢？已是就是这样写代码很容易出BUG，所以这里的让人觉得很绕的代码，就是很多的状态的判断，这些都是为了保证这个流程的安全。

### 5. 总结 ###

从部分源码的角度去分析，得到的结论是线程池也可能会导致OOM

那再思考一个问题：不断的创建线程池，“一定”会导致OOM吗？ 如果你对线程池已经有一定的了解，相信你也知道这个问题的答案。