vlambda博客
学习文章列表

一文读懂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


《源码系列》




《经典书籍》



《服务端技术栈》



《算法系列》






《设计模式》







扫描二维码

获取技术干货

后台技术汇




点个“在看”表示朕

已阅