本文主要改编自https://www.sohu.com/a/391767502_355142。下面从一道面试题引入本文主题~~
面试官:"假设有一个线程池,核心线程数为10,最大线程数为20,任务队列长度为100。如果现在来了100个任务,那么线程池里有几个线程在运行?"
粉丝豪:"应该是10吧!"
面试官:"你确定?"
粉丝豪:"确定啊!就是10…"
于是乎,漂亮的HR小姐姐让粉丝豪回去等通知了~
大家如果看出来了此题的陷阱,就不用看本文了!其实,这道题正确的答案是"不一定!"因为并没指明是哪一种线程池机制,带着这个疑问继续往下看!我们基于jdk 8,以两类线程池机制——先放队列再创建线程和先创建线程再放入队列——来剖析这道面试题。
先放队列再创建线程
核心线程数用完后,新来的任务先进队列,在队列满的时候,再创建线程。这种情况是大家最容易想到的情况,因为JDK中的线程池,也就是ThreadPoolExecutor就是这种机制!OK,我们先来看一下ThreadPoolExecutor的void execute(Runnable command)方法源码,如下图所示:
在int c = ctl.get()
代码上方,折叠了如下所示的一段英文注释,解释了上述截图中的三步流程:
/** Proceed in 3 steps:** 1. If fewer than corePoolSize threads are running, try to* start a new thread with the given command as its first* task. The call to addWorker atomically checks runState and* workerCount, and so prevents false alarms that would add* threads when it shouldn't, by returning false.** 2. If a task can be successfully queued, then we still need* to double-check whether we should have added a thread* (because existing ones died since last checking) or that* the pool shut down since entry into this method. So we* recheck state and if necessary roll back the enqueuing if* stopped, or start a new thread if there are none.** 3. If we cannot queue task, then we try to add a new* thread. If it fails, we know we are shut down or saturated* and so reject the task.*/
如果对英文不感冒,请参考下面的中文翻译:
- 判断当前活跃线程数是否小于corePoolSize,如果小于,则调用addWorker创建线程执行任务;
- 如果不小于corePoolSize,则将任务添加到workQueue队列;
- 如果放入workQueue失败,则创建线程执行任务,如果这时创建线程失败(当前线程数不小于maximumPoolSize时),就会调用函数reject拒绝接受任务。
用一张流程图来解释,如下:
如图所示,默认的机制为线程池里的核心线程数不够了,后面进来的任务会先丢队列,当队列满了,才起新线程。
因此,按照这套机制!粉丝豪的回答是正确的,当有100个任务添加进来时,先创建10个核心线程,剩下90个任务都丢进阻塞队列,因此线程池里只有10个线程在执行任务!
先创建线程再放入队列
当核心线程数用完后,如果来了新任务,则先创建线程,直至达到最大线程数,再把新任务放入阻塞队列。在dubbo中,有一种这种机制的线程池叫EagerThreadPoolExecutor线程池;在Tomcat里面也有类似的线程池。
来看一下EagerThreadPoolExecutor源码:
public class EagerThreadPoolExecutor extends ThreadPoolExecutor {/*** task count*/private final AtomicInteger submittedTaskCount = new AtomicInteger(0);public EagerThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit, TaskQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);}/*** @return current tasks which are executed*/public int getSubmittedTaskCount() {return submittedTaskCount.get();}@Overrideprotected void afterExecute(Runnable r, Throwable t) {submittedTaskCount.decrementAndGet();}@Overridepublic void execute(Runnable command) {if (command == null) {throw new NullPointerException();}// do not increment in method beforeExecute!submittedTaskCount.incrementAndGet();try {super.execute(command);} catch (RejectedExecutionException rx) {// retry to offer the task into queue.final TaskQueue queue = (TaskQueue) super.getQueue();try {//将任务提交到队列中if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {submittedTaskCount.decrementAndGet();throw new RejectedExecutionException("Queue capacity is full.", rx);}} catch (InterruptedException x) {submittedTaskCount.decrementAndGet();throw new RejectedExecutionException(x);}} catch (Throwable t) {// decrease any waysubmittedTaskCount.decrementAndGet();throw t;}}
}
主要重写了ThreadPoolExecutor的函数void execute(Runnable command),如果触发拒绝策略,那么将任务提交到TaskQueue阻塞队列中,再看TaskQueue源码:
public class TaskQueue<R extends Runnable> extends LinkedBlockingQueue<Runnable> {private static final long serialVersionUID = -2635853580887179627L;private EagerThreadPoolExecutor executor;public TaskQueue(int capacity) {super(capacity);}public void setExecutor(EagerThreadPoolExecutor exec) {executor = exec;}@Overridepublic boolean offer(Runnable runnable) {if (executor == null) {throw new RejectedExecutionException("The task queue does not have executor!");}int currentPoolThreadSize = executor.getPoolSize();// have free worker. put task into queue to let the worker deal with task.//如果提交任务数小于当前工作线程数,说明当前工作线程足够处理任务,将提交的任务插入到工作队列if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {return super.offer(runnable);}// return false to let executor create new worker.//重写代码的精髓之处:如果提交任务数大于当前工作线程数并且小于最大线程数,说明提交的任务量线程已经处理不过来,那么需要增加线程数,返回falseif (currentPoolThreadSize < executor.getMaximumPoolSize()) {return false;}// currentPoolThreadSize >= max//工作线程数到达最大线程数,插入到workqueuereturn super.offer(runnable);}/*** retry offer task** @param o task* @return offer success or not* @throws RejectedExecutionException if executor is terminated.*/public boolean retryOffer(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {if (executor.isShutdown()) {throw new RejectedExecutionException("Executor is shutdown!");}return super.offer(o, timeout, unit);}
}
主要重写了LinkedBlockingQueue的offer方法,而if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS))则调用offer,保证在当前线程数小于最大线程数时,直接新增线程。
因此,如果按照这么一套机制,粉丝豪的答案就不正确了。线程池启动完毕后进来100个任务时,直接会起20个线程,剩下的80个任务都会被丢进阻塞队列,综上所述,现在线程池里有20个线程在运行。
Reference
- https://blog.csdn.net/qq_26418435/article/details/102661092
- dubbo四种线程池源码分析