一文读懂JDK源码：ThreadPoolExecutor

Posted 后台技术汇

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了一文读懂JDK源码：ThreadPoolExecutor相关的知识，希望对你有一定的参考价值。


一文读懂JDK源码：ThreadPoolExecutor






线程池的思想是一种对象池的思想，开放一块内存空间，里面存放了众多（未死亡）的线程，池中线程执行调度由池管理器来处理。

当有线程任务时，从池中取一个工作线程并执行完任务单元，之后再把工作线程对象归还给池，从而避免反复创建线程对象所带来的性能开销，节省了系统的资源。

下面我们从四个角度出发，剖析“线程池”：

1.ThreadPoolExecutors的七个参数

2.Executors 源码分析

3.JDK线程池是如何完成工作调度呢？

4.线程池自定义配置案例



winter

开始之前，我们复习下 Executors 提供的五种线程池：

  • newCachedThreadPool 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
  • newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
  • newScheduledThreadPool 创建一个定长线程池，支持定时（scheduleWithFixedDelay（）函数的initdelay 参数）及周期（delay 参数）任务执行。
  • newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
  • newSingleThreadScheduledExecutor 创建一个单线程化的支持定时的线程池，可以用一个线程周期性执行任务(比如周期7天，一次任务才用1小时,使用多线程就会浪费资源)

参考下源码的方法列表：重载的方法都提供了一个 ThreadFactory（自定义线程工厂），我们通过 ThreadFactory 可以设置异步线程的异常处理等等。

一文读懂JDK源码：ThreadPoolExecutor


线程池生命周期有五个状态： 


 // 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;


一文读懂JDK源码：ThreadPoolExecutor


其生命周期转换如下图所示：


一文读懂JDK源码：ThreadPoolExecutor

一文读懂JDK源码：ThreadPoolExecutor




一文读懂JDK源码：ThreadPoolExecutor
ThreadPoolExecutors的七个参数






通过阅读源码，我们知道Executors的五个静态方法，底层最终都会创建一个 ThreadPoolExecutors对象：


//可以延期执行或者周期执行    ScheduledExecutorService executorService = Executors.newScheduledThreadPool(10);    //工作线程数量，基本大小=1，最大大小=1,FIFO ExecutorService executorService = Executors.newSingleThreadExecutor();    //线程池的工作线程数量是无界的,默认存活时间60s，超过会被kill掉,默认没有拒绝策略 ExecutorService executorService = Executors.newCachedThreadPool();//线程池的工作线程数量基础大小 = 数量最大值; 拒绝策略是超过了基础数据，则会抛异常 RejectedExecutionException。//线程存活时间，0，不会出现多余工作线程,自定义：线程工厂    ExecutorService executorService = Executors.newFixedThreadPool(10);    //单线程调度执行任务 ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();    


ThreadPoolExecutors 构造器：


public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {    //...     }


ThreadPoolExecutor 的构造器有7个入参配置，见下面参数列表：


参数
 定义
 作用
 备注
corePoolSize
 池子的基本容量
 长期驻留线程池的工作线程数量

allowCoreThreadTimeOut为true，该值为true，则线程池数量最后销毁到0个。
maximumPoolSize
 池子的最大容量
 定义池子最大容量
 allowCoreThreadTimeOut为false，会对超出基本容量的线程进行销毁，
销毁机制：超过核心线程数时，而且（超过最大值或者timeout超时），就会销毁。
keepAliveTime
  当线程池线程数量大于corePoolSize时候，多出来的空闲线程，多长时间会被销毁。
 必须大于0，默认是。0
unit
 生存时间的单位时间
 参考枚举类：
java.util.concurrent.TimeUnit
workQueue
 工作线程队列
 用于存放提交但是尚未被执行的任务
threadFactory
 线程工厂 用于创建线程
handler
 拒绝策略 指将任务添加到线程池中时，线程池拒绝该任务所采取的相应策略。




一文读懂JDK源码：ThreadPoolExecutor
Executors 源码分析







无界定时调度-线程池




我们且看第一个线程池：ScheduledExecutorService ；

ScheduledExecutorService executorService = Executors.newScheduledThreadPool(10);

最终构造一个 ThreadPoolExecutor 对象，它的构造器源码：

public class ScheduledThreadPoolExecutor extends ThreadPoolExecutorimplements ScheduledExecutorService {public ScheduledThreadPoolExecutor(int corePoolSize) {super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,new DelayedWorkQueue()); }}


代码分析：

  1. 线程池最大线程容量 maximumPoolSize = Integer.MAX_VALUE；
  2. 工作线程队列是 DelayedWorkQueue：它是一个优先级队列容器(肯定是优先级队列呀，延迟低的任务必须必延迟高的任务先被执行)，它保证添加到队列中的任务，会按照任务的延时时间进行排序，延时时间少的任务首先被获取；
  3. 超出基本大小的线程会被立即销毁，因此 keepAliveTime 设置为 0 纳秒了。


总结：

好处：利用优先级线程，确保了任务周期性或者带延迟的被执行，满足特点的业务需求；

弊端：由于最大线程池容量不设限，在提交任务极其频繁的条件下，有服务资源消耗殆尽的困难。


一文读懂JDK源码：ThreadPoolExecutor




单线程-线程池




我们且看第二个线程池：

ExecutorService executorService2 = Executors.newSingleThreadExecutor();

 

最终构造了一个 FinalizableDelegatedExecutorService 对象：ExecutorService 接口的 FinalizableDelegatedExecutorService 实现类（它是 Executors 的一个静态内部类）；

 static class FinalizableDelegatedExecutorService extends DelegatedExecutorService { FinalizableDelegatedExecutorService(ExecutorService executor) { super(executor); } protected void finalize() { super.shutdown(); }    }


Executors 的 newSingleThreadExecutor() 工具方法：

 public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()));    }


代码分析：

  1. 线程池最大线程容量 maximumPoolSize =1；
  2. 工作线程队列是 LinkedBlockingQueue：它是基于链表结构的有界阻塞队列，特点是FIFO；
  3. 超出基本大小的线程会被立即销毁，因此 keepAliveTime 设置为 0 毫秒了。


总结：

好处：阻塞工作队列，确保同时被执行的任务顺序串行执行，满足单线程执行任务的特定需求；如果线程池的唯一线程因为异常结束，那么会有一个新的线程来替代它；

弊端：一是假设先后提交的任务A和任务B，两者之间存在资源依赖（A依赖于B的执行结果），会导致线程池陷入死锁。

二是当添加任务的速度大于线程池处理任务的速度，可能会在队列堆积大量的请求，消耗很大的内存，甚至导致OOM。


一文读懂JDK源码：ThreadPoolExecutor




无界-线程池




我们且看第三个线程池：

ExecutorService executorService3 = Executors.newCachedThreadPool();

  

最终构造了一个ThreadPoolExecutor对象：

 public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());    }


代码分析：

  1. 线程池基本线程容量 corePoolSize = 0，也就是说池子里没有初始化好的线程资源；
  2. 线程池最大线程容量 maximumPoolSize = Integer.MAX_VALUE ;
  3. 工作线程队列是 SynchronousQueue：它是不存储元素的阻塞队列，每个插入操作都必须等待一个移出操作，否则一直put线程会一直阻塞（内部维护了一个Transferer 抽象类，提供了公平抢占消费&非公平抢占消费的实现）；
  4. 超出基本大小的线程资源在一段时间后会被销毁，因此 keepAliveTime 设置为 60 秒了。


总结：

好处：“无界限”的线程池，可以在资源被完全耗尽之前能够全力处理所有的任务提交（双刃剑）；

弊端：由于最大线程池容量不设限，在提交任务极其频繁的条件下，可能会创建数量非常多的线程，甚至OOM。


一文读懂JDK源码：ThreadPoolExecutor



有界-线程池




我们且看第四个线程池：

ExecutorService executorService4 = Executors.newFixedThreadPool(10);


最终构造了一个 ThreadPoolExecutor 对象：

   public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());    }


代码分析：
  1. 线程池基本线程容量  corePoolSize&maximumPoolSize  都是固定值，也就是说池子里一直维持一个固定数量的线程资源；
  2. 工作线程队列是  LinkedBlockingQueue ：它是基于链表结构的有界阻塞队列，特点是FIFO；
  3. 因为不允许超出固定大小的线程资源，因此 keepAliveTime 设置为 0 秒了。


总结：

好处：线程池的长度限制为固定的数值，确保。


一文读懂JDK源码：ThreadPoolExecutor




单线程-调度线程池




我们且看第五个线程池： 
 ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

  

最终构造了一个 DelegatedScheduledExecutorService 对象：它是 ExecutorService 接口的 FinalizableDelegatedExecutorService 实现类（是 Executors 的一个静态内部类）；

 public static ScheduledExecutorService newSingleThreadScheduledExecutor() { return new DelegatedScheduledExecutorService (new ScheduledThreadPoolExecutor(1)); } public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); }


 static class DelegatedScheduledExecutorService extends DelegatedExecutorService implements ScheduledExecutorService { private final ScheduledExecutorService e; DelegatedScheduledExecutorService(ScheduledExecutorService executor) { super(executor); e = executor; }  }


代码分析：

  1. 线程池基本线程容量 corePoolSize=1；
  2. 工作线程队列是 DelayedWorkQueue：它是一个优先级队列容器(肯定是优先级队列呀，延迟低的任务必须必延迟高的任务先被执行)，它保证添加到队列中的任务，会按照任务的延时时间进行排序，延时时间少的任务首先被获取；


总结：

好处：阻塞工作队列，确保同时被执行的任务顺序串行执行，满足单线程执行任务的特定需求；如果线程池的唯一线程因为异常结束，那么会有一个新的线程来替代它；

弊端：跟“无界调度线程池”一样，当添加任务的速度大于线程池处理任务的速度，可能会在队列堆积大量的请求，消耗很大的内存，甚至导致OOM。





一文读懂JDK源码：ThreadPoolExecutor
JDK线程池是如何完成工作调度呢？




那么一个线程池，最终是如何工作的呢？阻塞队列和工作线程又是怎么配合，实现快速消费任务呢？


任务调度




任务调度是线程池的主要入口，当用户提交了一个任务，接下来这个任务将如何执行都是由这个阶段决定的。了解这部分就相当于了解了线程池的核心运行机制。

首先，所有任务的调度都是由execute方法完成的，这部分完成的工作是：

  1. 检查现在线程池的运行状态、运行线程数、运行策略；
  2. 决定接下来执行的流程，是直接申请线程执行，或是缓冲到队列中执行，亦或是直接拒绝该任务。


我们通过一张图来理解下：


一文读懂JDK源码：ThreadPoolExecutor



  • A 首先检测线程池运行状态，如果不是RUNNING，则直接拒绝，线程池要保证在RUNNING的状态下执行任务。

  • B 如果 workerCount < corePoolSize，则创建并启动一个线程来执行新提交的任务。（基本大小线程数量没凑够，得加人手..）

  • C 如果 workerCount >= corePoolSize，且线程池内的阻塞队列未满（不阻塞），则将任务添加到该阻塞队列中。（基本大小满足了，还有临时工也在帮忙，再来单子得阻塞..）

  • D 如果 workerCount >= corePoolSize &&

    workerCount < maximumPoolSize，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务。（基本大小的干活人数凑够了，临时人数，而且单子又堆满了，那只能在限制最大人数前提下，继续招临时工来帮忙了..）

  • E 如果 workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。（厂子就这麽大，基本大小的干活人，加上临时工，单子排的满满的，再来订单我们不接了..）


通过逻辑，我们可以理解源码：ThreadPoolExecutor.execute(Runnable command) 
 public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get();        // B -  workerCount < corePoolSize if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); }        // C - workerCount >= corePoolSize，且线程池内的阻塞队列未满 if (isRunning(c) && workQueue.offer(command)) {            int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command);            else if (workerCountOf(recheck) == 0)             addWorker(null, false);        }        // D - workerCount >= corePoolSize && workerCount < maximumPoolSize，且线程池内的阻塞队列已满 else if (!addWorker(command, false))            // E - workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满 reject(command); }


补充：

以上源码是线程池的任务调度逻辑，此外“任务调度”还涉及了线程池的任务申请、任务拒绝，篇幅所限，这里不展开讲解了。所以，推荐一篇精品文章给大家自行阅读：《Java线程池实现原理及其在美团业务中的实践》。




一文读懂JDK源码：ThreadPoolExecutor
线程池自定义配置案例




阿里规约提倡手动创建线程池，而非Java内置的线程池：“ 线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。”
通过上面我们分析了 Executors 的多个工具方法方法，最终发现底层都是依赖于创建 ThreadPoolExecutor 线程池，并且我们知道 ThreadPoolExecutor 的关键配置项有 7 个：corePoolSize、maximumPoolSize、keepAliveTime、unit、workQueue、threadFactory、defaultHandler。
一文读懂JDK源码：ThreadPoolExecutor




这里提供一个代码实现的案例：
1、将线程池对象封装到一个工具类里面，Util工具类封装一个提交任务的api
2、通过 完成线程池的构造（比较符合一般访问量的服务能力了）

  1. 设置线程池核心线程数量为5，
  2. 线程池是150个最大线程量，
  3. 等待执行任务队列长度最大为150个任务，
  4. ArrayBlockingQueue 作为任务队列，
  5. 超出线程池部分的资源，则保持1800s的存活时间（半小时）

ExecutorUtil.java

/** * 线程池,任务调度工具类 * */public final class ExecutorUtil { /** * 线程池 */ private static ExecutorService threadpool = ThreadUtil.newExecutorService(5, 150, 150, 1800, "test-executors");       /** * 执行任务 * @param task - 任务 * @return - 执行期望值 */ public static Future<?> submit(Runnable task) { return threadpool.submit(task);  }}


ThreadUtil.java

/** * 线程池工厂类
 */public final class ThreadUtil {
 /** * 根据参数创建执行者服务 * @param coreSize -- 线程池核心线程数 * @param maxSize -- 线程池最大线程数 * @param queueSize -- 线程池等待队列长度 * @param keepAlive -- 线程最大空闲时间(单位:秒) * @param nameTemplate -- 线程名称模板 * @return -- 执行者服务 */ public static ExecutorService newExecutorService(int coreSize,  int maxSize, int queueSize, int keepAlive, final String nameTemplate) { BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(queueSize); final ThreadGroup tg = new ThreadGroup(nameTemplate); tg.setDaemon(true);    ThreadFactory fac = new ThreadFactory() {             private int index = 0;       // 创建一个新的线程, 同时设置它的名称和daemon模式 @Override      public Thread newThread(Runnable r) {         long stackSize = 256 * 1024; String tn = nameTemplate + "_" + index++;  Thread t = new Thread(tg, r, tn, stackSize); t.setDaemon(true); return t; }  };  ThreadPoolExecutor tp = new ThreadPoolExecutor(coreSize, maxSize, keepAlive, TimeUnit.SECONDS, queue, fac); tp.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 当达到阀值后使用当前调用线程执行任务 return tp; }}





一文读懂JDK源码：ThreadPoolExecutor
总结




至此，我们完成了对线程池的四个角度的剖析，分别是：

1.ThreadPoolExecutors的七个参数

2.Executors 源码分析

3.JDK线程池是如何完成工作调度呢？

4.线程池自定义配置案例

文章篇幅有限，对某些线程池细节的点可能还有遗漏，大家可以对照思路，参考阅读线程池的相关源码，或者下面的文章参考列表，这样可以加深大家对“线程池”的理解。希望内容对大家有所帮助，晚安~~

文章参考：
https://www.cnblogs.com/thisiswhy/p/12782548.html （每天都在用，但你知道 Tomcat 的线程池有多努力吗）
https://juejin.cn/post/6844904122760560648（如何设置线程池参数？美团给出了一个让面试官虎躯一震的回答）
https://tech.meituan.com/2020/04/02/java-pooling-pratice-in-meituan.html（Java线程池实现原理及其在美团业务中的实践）




往期推荐


一文读懂JDK源码：ThreadPoolExecutor


《源码系列》

《》

《》

《》

《》

《》


《经典书籍》

《》

《》

《》


《服务端技术栈》

》



《算法系列》

《》





《设计模式》

《》

《》

《》

《》

《》

《》

《》






扫描二维码

获取技术干货

后台技术汇




点个“在看”表示朕

已阅



以上是关于一文读懂JDK源码：ThreadPoolExecutor的主要内容，如果未能解决你的问题，请参考以下文章

从源码入手，一文带你读懂Spring AOP面向切面编程

从源码入手，一文带你读懂Spring AOP面向切面编程

面试必备：从源码入手，带你一文读懂Spring AOP面向切面编程

一文读懂JDK7,8,JD9的hashmap，hashtable，concurrenthashmap及他们的区别

一文读懂深克隆与浅克隆的关系

一文读懂CentOS 8 安装JDK 11：配置JAVA_HOME环境变量