导读:线程池是开发中使用频率比较高的组件之一,但是又有多少人真正了解其内部机制呢。

关键词:线程池

前言

线程池是大家开发过程中使用频率比较高的组件之一,但是其内部原理又有多少人真正清楚呢。最近抽时间去了解了一下其内部实现细节,感觉略有收获,遂以ThreadPoolExecuter为例将自己的心得体会分享出来和大家一起交流,如有不妥之处,烦请大家积极指正。

我的疑问

  1. 执行任务的工作机制是怎样的?
  2. 是什么时机清理空闲线程的?
  3. ThreadPoolExecutor.CallerRunsPolicy 什么场景下会用到?
  4. 线程池终止的时候都做了什么?(过两天补上)

问题1-执行任务的工作机制是怎样的?

开始之前,我先试图用一个真实的场景来描述一下,2015年我们公司boss开始创立公司,创立初期公司业务比较少,他一个人(corePoolSize=1)干的有条不紊,有声有色的,没过多久,业务量上来了,他一个人干不过来,分身乏力,那怎么办呢?其实很简单,排队呗,就这样boss将待办的任务都添加到备忘录(BlockIngQueue)里面,boss又开始愉快的工作,但是客户的耐心终归有限,过了几天发现自己交给我们公司的业务还没完成,客户一气之下打电话给boss“我的活你干完没有,没干的话就停下来(shutdown/shutdownNow)吧,我找别人了”,这时候boss慌了,默默的点上一根烟,在网上发了招聘通知,就这样干活的人又多了起来(addWorker),公司在boss的带领下风生水起,就这样不知不觉中公司走过了五个年头,本以为可以大干一场的我们,却偏偏赶上了2020年的新冠,复工日期一拖再拖,客户需求一少再少,唯独公司养的员工没少,这是公司目前最大的开支了。长痛不如短痛,boss下发了一个政策,如果员工本月(keepAliveTime=一个月)kpi完不成,假定kpi为“单月完成任务数大于0“,那么就会被淘汰(空闲线程被清理),最后撑了两个月,公司又回到解放前,boss又成了光杆司令。

文字描述

提交任务步骤1:启动核心工作线程

触发条件:工作线程数小于核心线程数

步骤描述:启动核心工作线程,并将任务作为工作线程的firstTask,如果启动失败会走到“提交任务步骤2”

提交任务步骤2:往阻塞队列中堆积任务

触发条件:工作线程数大于核心线程数

步骤描述:尝试将任务堆积到队列中,如果堆积失败,会走到“提交任务步骤3”。如果堆积成功会做两个双重检查,分别是状态的检查和工作线程数的检查,如果状态不合法就尝试删除刚添加成功的任务,删除成功调用reject方法(为什么删除失败不需要调用reject方法呢?因为删除失败意味着task已经被处理过了,不能谎报军情);如果工作线程数等于0就补充工作线程,什么情况下工作线程会变成0呢,后面会单独说这个问题

提交任务步骤3:启动工作线程

触发条件:核心线程数已达到线程池设定的阈值corePoolSize而且队列里已经堆积不了了

步骤描述:尝试启动工作线程,如果这时候启动失败就调用reject方法通知调用者

时序图描述

java ThreadPoolExecutor初探-LMLPHP

问题2-是什么时机清理空闲线程的?

我们先来回顾下线程的运行状态,如下图:

java ThreadPoolExecutor初探-LMLPHP

回到我们的问题,只需要让“空闲线程真的空闲下来”它自然就被清理了,那怎么能让它真的闲下来呢。前面时序图的第5步我提到worker会一直循环从BlockingQueue里面获取task执行,如果没有task返回给worker,那说明它真的是闲了(getTask返回null,意味着会退出while循环,很快worker的run方法就执行结束,线程从RUNNABLE到TERMINATED),它需要自己退出历史的长河中了,纵使它曾经立下汗马功劳,但是谁让它不是核心人员呢(你处于corePoolSize之外),让我们读一下getTask的代码:

private Runnable getTask() {
//标记位,获取任务是否超时
boolean timedOut = false; // Did the last poll() time out? for (;;) { int wc = workerCountOf(c);//worker的数量 // 标记位,可以简单理解为是否允许超时回收,允许的条件为allowCoreThreadTimeOut(是否允许核心线程被回收)或者worker的数量大于核心线程数
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; //满足以下几个条件时会返回null,1.worker的数量大于最大线程池的数量;
2.允许超时回收(timed==true)&&上一次获取任务超时(timedOut==true)&&当前没有堆积任务(workQueue.isEmpty),说明worker确实空闲了
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
} try {
//如果允许超时回收就使用poll,否则使用take
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take(); if (r != null)
return r;
//poll结束,队列中当时没有任务,将timedOut置为true,下一趟循环时会满足回收的第二个条件,getTask返回null timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}

  

问题3-ThreadPoolExecutor.CallerRunsPolicy 什么场景下会用到?

前面我们提到,如果线程池的处理能力已经饱和,那么就会调用reject方法执行RejectedExecutionHandler,jdk已经为我们提供了四种拒绝策略,分别是AbortPolicy,DiscardOldestPolicy,DiscardPolicy,CallerRunsPolicy,前三种都好理解,简单来说就是丢弃,唯独最后一种我想了好久没想到使用场景,前不久在看同事代码的时候发现他有用到这个策略,来达到一种限流的效果,简单画个流程图说明一下逻辑:

DelayQueuePollingTask负责定时从数据库拉取延迟队列,然后封装成task扔到线程池去执行,task使用httpclient向isv推送数据,由于isv的服务吞吐量比较低,经常会触发超时,进而导致线程池被跑满,为了让DelayQueuePollingTask可以感知到线程池处于饱和状态,而且又不至于它空等待,所以使用了CallerRunsPolicy这个策略,当线程池满负荷的时候由DelayQueuePollingTask所在的线程负责执行task。

java ThreadPoolExecutor初探-LMLPHP

总结

上周末带着自己之前使用线程池过程中的一些疑问学习了一下ThreadPoolExecutor源码,虽说没有完全吃透,但也带给自己不少收获,在这里将自己的思考过程分享出来,但愿能帮到一部分人。

如果觉得有用,请点个推荐。

05-18 02:22