写点什么

Java 线程池原理分析,java 项目经理面试常见问题及答案

用户头像
极客good
关注
发布于: 刚刚

| 2 | 线程数 ≥?corePoolSize,且 workQueue 未满 | 缓存新任务 |


| 3 | corePoolSize ≤ 线程数 <?maximumPoolSize,且?workQueue 已满 | 创建新线程 |


| 4 | 线程数 ≥?maximumPoolSize,且 workQueue 已满 | 使用拒绝策略处理 |

3.1.3 资源回收

考虑到系统资源是有限的,对于线程池超出 corePoolSize 数量的空闲线程应进行回收操作。进行此操作存在一个问题,即回收时机。目前的实现方式是当线程空闲时间超过 keepAliveTime 后,进行回收。除了核心线程数之外的线程可以进行回收,核心线程内的空闲线程也可以进行回收。回收的前提是allowCoreThreadTimeOut属性被设置为 true,通过public void allowCoreThreadTimeOut(boolean) 方法可以设置属性值。

3.1.4 排队策略

如 3.1.2 线程创建规则一节中规则 2 所说,当线程数量大于等于 corePoolSize,workQueue 未满时,则缓存新任务。这里要考虑使用什么类型的容器缓存新任务,通过 JDK 文档介绍,我们可知道有 3 中类型的容器可供使用,分别是同步队列有界队列无界队列。对于有优先级的任务,这里还可以增加优先级队列。以上所介绍的 4 中类型的队列,对应的实现类如下:


| 实现类 | 类型 | 说明 |


| --- | --- | --- |


| SynchronousQueue | 同步队列 | 该队列不存储元素,每个插入操作必须等待另一个线程调用移除操作,否则插入操作会一直阻塞 |


| ArrayBlockingQueue | 有界队列 | 基于数组的阻塞队列,按照 FIFO 原则对元素进行排序 |


| LinkedBlockingQueue | 无界队列 | 基于链表的阻塞队列,按照 FIFO 原则对元素进行排序 |


| PriorityBlockingQueue | 优先级队列 | 具有优先级的阻塞队列 |

3.1.5 拒绝策略

如 3.1.2 线程创建规则一节中规则 4 所说,线程数量大于等于 maximumPoolSize,且 workQueue 已满,则使用拒绝策略处理新任务。Java 线程池提供了 4 中拒绝策略实现类,如下:


| 实现类 | 说明 |


| --- | --- |


| AbortPolicy | 丢弃新任务,并抛出?RejectedExecutionException |


| DiscardPolicy | 不做任何操作,直接丢弃新任务 |


| DiscardOldestPolicy | 丢弃队列队首的元素,并执行新任务 |


| CallerRunsPolicy | 由调用线程执行新任务 |


以上 4 个拒绝策略中,AbortPolicy 是线程池实现类所使用的策略。我们也可以通过方法public void setRejectedExecutionHandler(RejectedExecutionHandler)修改线程池决绝策略。

3.2 重要操作

3.2.1 线程的创建与复用

在线程池的实现上,线程的创建是通过线程工厂接口ThreadFactory的实现类来完成的。默认情况下,线程池使用Executors.defaultThreadFactory()方法返回的线程工厂实现类。当然,我们也可以通过


public void setThreadFactory(ThreadFactory)方法进行动态修改。具体细节这里就不多说了,并不复杂,大家可以自己去看下源码。


在线程池中,线程的复用是线程池的关键所在。这就要求线程在执行完一个任务后,不能立即退出。对应到具体实现上,工作线程在执行完一个任务后,会再次到任务队列获取新的任务。如果任务队列中没有任务,且 keepAliveTime 也未被设置,工作线程则会被一致阻塞下去。通过这种方式即可实现线程复用。


说完原理,再来看看线程的创建和复用的相关代码(基于 JDK 1.8),如下:


`+----ThreadPoolExecutor.Worker.java Worker(Runnable firstTask) {


setState(-1);


this.firstTask = firstTask;


// 调用线程工厂创建线程


this.thread = getThreadFactory().newThread(this);


}


// Worker 实现了 Runnable 接口


public void run() {


runWorker(this);


}


+----ThreadPoolExecutor.java final void runWorker(Worker w) {


Thread wt = Thread.currentThread();


Runnable task = w.firstTask;


w.firstTask = null;


w.unlock();


boolean completedAbruptly = true;


try {


// 循环从任务队列中获取新任务


while (task != null || (task = getTask()) != null) {


w.lock();


// If pool is stopping, ensure thread is interrupted;


// if not, ensure thread is not interrupted. This


// requires a recheck in second case to deal with


// shutdownNow race while clearing interrupt


if ((runStateAtLeast(ctl.get(), STOP) ||


(Thread.interrupted() &&


runStateAtLeast(ctl.get(), STOP))) &&


!wt.isInterrupted())


wt.interrupt();


try {


beforeExecute(wt, task);


Throwable thrown = null;


try {


// 执行新任务


task.run();


} catch (RuntimeException x) {


thrown = x; throw x;


} catch (Error x) {


thrown = x; throw x;


} catch (Throwable x) {


thrown = x; throw new Error(x);


} finally {


afterExecute(task, thrown);


}


} finally {


task = null;


w.completedTasks++;


w.unlock();


}


}


completedAbruptly = false;


} finally {


// 线程退出后,进行后续处理


processWorkerExit(w, completedAbruptly);


}


}`

3.2.2 提交任务

通常情况下,我们可以通过线程池的submit方法提交任务。被提交的任务可能会立即执行,也可能会被缓存或者被拒绝。任务的处理流程如下图所示:



上面的流程图不是很复杂,下面再来看看流程图对应的代码,如下:


`+---- AbstractExecutorService.java


public Future<?> submit(Runnable task) {


if (task == null) throw new NullPointerException();


// 创建任务


RunnableFuture<Void> ftask = newTaskFor(task, null);


// 提交任务


execute(ftask);


return ftask;


}


+---- ThreadPoolExecutor.java public void execute(Runnable command) {


if (command == null)


throw new NullPointerException();


int c = ctl.get();


// 如果工作线程数量 < 核心线程数,则创建新线程


if (workerCountOf(c) < corePoolSize) {


// 添加


【一线大厂Java面试题解析+核心总结学习笔记+最新架构讲解视频+实战项目源码讲义】
浏览器打开:qq.cn.hn/FTf 免费领取
复制代码


工作者对象


if (addWorker(command, true))


return;


c = ctl.get();


}


// 缓存任务,如果队列已满,则 offer 方法返回 false。否则,offer 返回 true


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


}


// 添加工作者对象,并在 addWorker 方法中检测线程数是否小于最大线程数


else if (!addWorker(command, false))


// 线程数 >= 最大线程数,使用拒绝策略处理任务


reject(command);


}


private boolean addWorker(Runnable firstTask, boolean core) {


retry:


for (;;) {


int c = ctl.get();


int rs = runStateOf(c);


// Check if queue empty only if necessary.


if (rs >= SHUTDOWN &&


! (rs == SHUTDOWN &&


firstTask == null &&


! workQueue.isEmpty()))


return false;


for (;;) {


int wc = workerCountOf(c);


// 检测工作线程数与核心线程数或最大线程数的关系


if (wc >= CAPACITY ||


wc >= (core ? corePoolSize : maximumPoolSize))


return false;


if (compareAndIncrementWorkerCount(c))


break retry;


c = ctl.get(); // Re-read ctl


if (runStateOf(c) != rs)


continue retry;


// else CAS failed due to workerCount change; retry inner loop


}


}


boolean workerStarted = false;


boolean workerAdded = false;


Worker w = null;


try {


// 创建工作者对象,细节参考上一节所贴代码


w = new Worker(firstTask);


final Thread t = w.thread;


if (t != null) {


final ReentrantLock mainLock = this.mainLock;


mainLock.lock();


try {


int rs = runStateOf(ctl.get());

用户头像

极客good

关注

还未添加个人签名 2021.03.18 加入

还未添加个人简介

评论

发布
暂无评论
Java 线程池原理分析,java项目经理面试常见问题及答案