【并发进阶】4000千字带你全面解析Java线程池原理

本是古典何须时尚 2024-04-08 13:53 23阅读 0赞

关于线程池使用方法的文章太多了，这里就不多啰嗦了，今天我们来聊细节，我知道大家对于如何使用线程池肯定比我熟悉，但是线程池创建流程的几个关键节点的策略你知道吗？比如第一次启动时候，核心线程数的创建是创建满还是先复用？空闲线程是如何释放的？线程池如何关闭？等，具体有如下几个问题：

1.  核心线程的创建策略是什么？比如核心线程数设置为5，我先提交一个任务，执行完后又提交一个任务，这个时候线程池里面有几个核心线程？是2个还是1个呢？
2.  当队列满的时候提交任务，会创建最大核心线程数相关线程，那么这个线程是从队列里面取还是直接用这个刚提交的任务呢？
3.  线程池的拒绝时机是什么？
4.  线程池是如何实现线程的重复利用的?
5.  空闲线程是如何释放的？
6.  核心线程是否能够回收?

> **良心提醒：** 涉及到源码，可能需要时间比较长，对于以上几个问题如果你都比较清楚，就不用往下看了，避免浪费时间。如果没时间可以先收藏哦

### 线程池概述 ###

我们先简单的复习一下线程池参数及流程

#### 线程池的6个参数 ####

![cc05f1bb4d3e43199af1ca78eb65be8e.png][]

首先，我们先来看下线程池中各个参数的含义，如表所示线程池主要有 6 个参数，其中第 3 个参数由 `keepAliveTime + 时间单位`组成。`corePoolSize` 是核心线程数，也就是常驻线程池的线程数量，与它对应的是 `maximumPoolSize`表示线程池最大线程数量，**当我们的任务特别多而 corePoolSize 核心线程数无法满足需求并且任务队列存放满的时候**，就会向线程池中增加线程，以便应对任务突增的情况。

#### 线程的创建流程 ####

![f1c355726ba54e45a52dcf11a250a6da.png][]

如上图所示，当提交任务后，线程池首先会检查当前线程数，如果此时线程数小于核心线程数，则新建线程并执行任务，随着任务的不断增加，线程数会逐渐增加并达到核心线程数，此时如果仍有任务被不断提交，就会被放入 workQueue 任务队列中，等待核心线程执行完当前任务后重新从 workQueue 中提取正在等待被执行的任务。

此时，如果我们的任务特别多，达到了workQueue的容量上限，线程池就会继续创建非核心线程来执行任务也就是maximumPoolSize控制的最大线程数，假设任务仍然不断提交，线程池会继续创建非核心线程来执行任务，当线程数达到maximumPoolSize规定的上限时，线程池就会拒绝这些任务，也就是执行线程的拒绝策略。

线程的整体创建流程如上面的描述，但是创建的细节我们需要结合源码来分析了。

### 线程池核心分析 ###

#### 线程池组成及功能 ####

![c2e3c277864e4080bad069e61792e6bc.png][]

如图，我们思考一下，如果我们自己实现线程池会怎么实现，顾名思义，线程池就是存放线程的池子，为了达到复用、资源控制等效果，**首先我们需要一个线程的生产工厂**，**其次我们需要一个管理线程的地方来管理线程的创建、执行、复用、销毁、拒绝等功能**，**然后还要个能排队的任务队列来排队等待执行任务**，**最后我们还需要实现不同的拒绝策略来处理超出线程城池自身处理能力的任务**

*  **ThreadFactory:** 线程的生产工厂，通过方法`Thread newThread(Runnable r)`来创建线程。
 *  **ThreadPoolExecutor**: 线程池的实现类用来管理线程池的创建、执行、复用、销毁、拒绝等功能。
 *  **BlockingQueue workQueue:** 存放任务的队列。
 *  **RejectedExecutionHandler:** 拒绝策略接口，当线程及任务满的时候，拒绝任务提交的策略。
 *  **Worker:** 复用的工作线程，存放到`ThreadPoolExecutor`中的 `HashSet<Worker>`里面。

带着上面的基本概念，我们来结合源码深入的分析一下

#### 1. 启动源码分析 ####

public static void main(String[] args) throws InterruptedException {
        //创建线程池 核心线程数为2，最大线程数为3的线程池
        ThreadPoolExecutor executorService = new ThreadPoolExecutor(
                2, 			// corePoolSize
                3, 			// maximumPoolSize
                60L,		// keepAliveTime
                TimeUnit.SECONDS, //单位
                new LinkedBlockingDeque<>(3), // workQueue
                Executors.defaultThreadFactory()); // threadFactory
        //线程池提交一个任务并执行
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("线程池创建线程");
            }
        });
    }
    复制代码

我们创建一个参数如代码所示的线程池，并提交一个 `Runnable` 任务，接下来我们看一下submit对应的源码：

public Future<?> submit(Runnable task) {
        RunnableFuture<Void> ftask = new FutureTask<T>(task, null);
        execute(ftask);
        return ftask;
    }
    // 执行线程的方法
    public void execute(Runnable command) {
        // 当前线程池线程的数量
        int c = ctl.get();
      	// 1. 获取当前线程池的线程数，如果小于核心线程数，创建核心线程 `addWorker(command, true)`
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
        }
      	// 2. 如果当前线程池是运行状态并且工作线程数量大于等于核心线程数，把任务添加到队列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
          	// 4. 如果当前线程池没有在运行(线程池调用shutdown()方法了)，直接清除任务并执行拒绝策略
            if (! isRunning(recheck) && remove(command))
                reject(command);
          	// 5. 否则如果线程池在运行并且工作线程数量为0，则创建非核心线程(比如 corePoolSize设置为0的时候，会走这里直接创建非核心线程)
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
      	/* 3. 否则这里分两种情况，
      		第一（isRunning(c)==false）: 那么addWorker() 会返回 false 直接执行拒绝策略
      		第二 (workQueue.offer()==false): 那么说明队列满了，这个时候 addWorker() 会返回 true
      	*/
        else if (!addWorker(command, false))
            reject(command);
    }
    复制代码

**我们看一下上面的代码步骤，正常流程创建 `1、2、3`**

1.  如果小于核心线程数，创建核心线程 `addWorker(command, true)`
2.  如果核心线程数满了，就把任务添加到队列
3.  如果队列满了，才会创建非核心线程。

基于这个正常的流程，我们来回顾一下开篇提到的`问题1，问题2`

**`问题1：`** 由代码可知，**只要工作线程数小于核心线程数**，不管工作线程是否有空，都会直接创建工作线程并直接任务。

**`问题2：`** 当队列满的时候，会直接`addWorker(command,false)`创建非核心线程并把任务传给它直接执行。

**我们再来看一下`问题3`拒绝策略触发的时机，我们结合流程`1、2、3`，`1、2、4`来看看。**

**`流程1、2、3`**：这个就是当工作线程数最大并且队列满的时候，添加线程会失败，触发拒绝策略`reject(command)`

**`流程1、2、3或4`**：如果线程池在运行的时候，突然有人调用线程池的shutdown方法了，这个时候 `isRunning(c)==false`就会触发拒绝策略，或者在进入`4` 之前shutdown了，那么会进入`3`这时`addWorker(command, false)会返回false`,也会触发拒绝策略。

> 拒绝策略触发时机小结:
> 
> 1.  线程池满了，任务队列满了会触发拒绝策略。
> 2.  当线程池在调用shutdown方法的时候，如果继续添加任务也会触发拒绝策略。

#### 2. Worker线程复用源码分析 ####

在上面启动源码的时候我们分析了，如果任务多了把任务放到`BlockingQueue`队列里面，其实复用就是工作线程一直从队列里面取任务，然后执行。我们来看一下 `addWorker()`方法的源码

private boolean addWorker(Runnable firstTask, boolean core) {
        ...
        boolean workerStarted = false;
        boolean workerAdded = false;
        ThreadPoolExecutor.Worker w = null;
        try {
          	// 1. 创建Worker 线程，这里注意点：在Worker里面的线程是通过我们最开始传入的线程工厂创建的
            w = new Worker(firstTask);
            final Thread t = w.thread;
          	... //这里省略其他逻辑
            workers.add(w);
          	workerAdded = true;
          	if (workerAdded) {
              // 2. 启动线程, 这里的start() 就是线程启动，我们看Worker类里面把this传递给thread了
              // 所以这里start()其实调用的是 Worker里面的 run() 方法。
              t.start();
              workerStarted = true;
            }
          	...
        return workerStarted;
    }
    
    class Worker extends AbstractQueuedSynchronizer implements Runnable
        // Worker类的构造方法, 并且Worker是 实现的 Runnable接口
        Worker(Runnable firstTask) {
            this.firstTask = firstTask;
            // 这里其实 <=> this.thread = new Thread(this);
            this.thread = getThreadFactory().newThread(this);
        }
      	// 3. 接着上面第2步 start() 方法过来
      	public void run() {
        		runWorker(this);
    		}
      	final void runWorker(Worker w) {
            Thread wt = Thread.currentThread();
          	// 4. 拿到真正的任务
            Runnable task = w.firstTask;
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try {
              	// 5. 这里如果task为null, 就从 队列里面循环获取，这里就是通过 getTask() 来获取的
                while (task != null || (task = getTask()) != null) {
                    w.lock();
                    // 这里相应线程中断，或者线程停止
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() &&
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                        wt.interrupt();
                    try {
                        try {
                          	// 6. 启动任务
                            task.run();
                        } catch (Throwable ex) {
                            throw ex;
                        }
                    } finally {
                        task = null;
                        w.unlock();
                    }
                }
              	// 工作线程中断或者异常跳出，会触发Worker线程回收工作
                completedAbruptly = false;
            } finally {
              	// 释放Worker线程，具体源码可自行查阅
                processWorkerExit(w, completedAbruptly);
            }
        }
    }
    复制代码

我们看了上面的代码，省去了大部分的非主流程的代码，我们可以看到，创建Worker线程的时候, **线程工厂会把Worder 自身传递给`Thread()`**, 这样在第`2`步的时候，`t.start()` 启动线程就会**异步触发 Worker中的 `run()` 方法**，在`runWorker()`方法中有个 **`while循环`一直的从队列中获取任务并运行**，**这里其实就是线程池中线程在重复的取任务执行任务，如此循环往复**

这个流程也解释了最开始的 `问题4`已经清晰了。

> 对于这部分代码t.start() 启动为啥调用 run()方法可以看一下我之前的文章。 [线程实现原理][Link 1]及[Runnable原理分析][Runnable]等文章

#### 3. 空闲线程是如何释放的？ ####

至此线程池创建，复用，拒绝都聊过，最后我们来看一下非工作线程是怎么释放的呢，在**Worker线程复用源码分析**中我们看到，`runWorker()`方法中有个while循环在一直循环的获取任务执行任务，当跳出循环的时候，会执行 `processWorkerExit(w, completedAbruptly)`方法，这个方法就是释放线程的，但是跳出循环的时机是什么呢？

上面代码我们可以看到，**当Worker线程被中断或者状态为STOP的时候会跳出**，这两个是线程池调用shutdown方法的时候触发的，咱这里先不考虑这个情况，除了这种情况还有个条件在`while((task = getTask()) != null)` 中，就是说获取不到`getTask()` 的时候也会跳出循环释放线程，那么我们来看看`getTask()`的源码

private Runnable getTask() {
      	//超时标志
        boolean timedOut = false;
      	// 循环
        for (;;) {
            int c = ctl.get();
            //1. 线程池状态为 SHUTDOWN并且队列为空或者STOP的时候会返回 null, 释放当前线程
            if (runStateAtLeast(c, SHUTDOWN)
                    && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
              	...
                return null;
            }
    				// 2. 获取当前线程池中线程的数量
            int wc = workerCountOf(c);
    
            // 3. 当允许核心线程超时或者 当前线程数量大于核心线程数时 timed = true
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    				// 4. timed == true 并且 timeOut == true 并且 队列任务为空且线程池线程存在的情况下，返回null, 释放当前线程
            if ((wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) {
              	// Worker 线程数减一操作
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
    
            try {
              	// 5. 这里是关键：timed==true 意思是线程数大于核心线程数的时候，从队列里面取值并加了个 keepAliveTime 超时时间，如果超过这个时间还没取到任务，就timedOut=true, 然后再次循环的时候，上面第4步的if 条件就满足了，就会return null, 然后就会释放线程了。
                Runnable r = timed ?
                        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                        workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }
    复制代码

其实到这一步我们可以看出，释放线程的原理是，上面代码的`第5步`，也就是**timed==true 意思是线程数大于核心线程数的时候，从队列里面取值并加了个 keepAliveTime 超时时间，如果超过这个时间还没取到任务，就timedOut=true, 然后再次循环的时候，上面第4步的if 条件就满足了，就会return null, 然后就会释放线程了。**

这里还有个点是，**allowCoreThreadTimeOut**这个参数来控制是否能释放核心线程数，默认是 **false**,可以通过 `allowCoreThreadTimeOut(boolean value)`方法来设置值,如果设置为`true`,意味着上面代码的`第3步`不会判断**当前线程数>核心线程数**这个条件，也就是说，线程池中只要有线程就可以释放。

至此对于最开始的 **`问题5，问题6`** 已经清晰了。

### 最后 ###

感谢各位能看到最后，希望本篇的内容对你有帮助，有什么意见或者建议可以留言一起讨论，看到后第一时间回复，也希望大家能给个赞，你的赞就是我写文章的动力，再次感谢。

> 最近面试的小伙伴很多，对此我整理了一份Java面试题手册：基础知识、JavaOOP、Java集合/泛型面试题、  
> Java异常面试题、Java中的IO与NIO面试题、Java反射、Java序列化、Java注解、多线程&并发、JVM、Mysql、Redis、  
> Memcached、MongoDB、Spring、SpringBoot、SpringCloud、RabbitMQ、Dubbo、MyBatis、ZooKeeper、数据结构、算法、  
> Elasticsearch、Kafka、微服务、Linux等等。可以分享给大家学习。【持续更新中】领取方式【999】就可以领取资料了

[cc05f1bb4d3e43199af1ca78eb65be8e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/7158e62bd8b541c88e676830d3b270ac.png
[f1c355726ba54e45a52dcf11a250a6da.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/dbe8102058ff4fbcb3caa07e67aa96fd.png
[c2e3c277864e4080bad069e61792e6bc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/1a2a6d1d9f654dab927d8f0ef84a9ccd.png
[Link 1]: https://juejin.cn/post/7116634679414882334
[Runnable]: https://juejin.cn/post/7114155426836709390