Java线程池原理分析
作者:互联网
1. 简介
Java中的线程池是运用场景最多的并发框架,几乎所有需要异步或并发执行任务的程序都可以使用线程池。在开发过程中,合理地使用线程池能够带来3个好处。
- 降低资源消耗 。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
- 提高响应速度 。当任务到达时,任务可以不需要等到线程创建就能立即执行。
- 提高线程的可管理性 。线程是稀缺资源,如果无限制地创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一分配、调优和监控。但是,要做到合理利用线程池,必须对其实现原理了如指掌。
2. 线程池的继承体系
如上图,最顶层的接口 Executor 仅声明了一个方法execute。ExecutorService 接口在其父类接口基础上,声明了包含但不限于shutdown、submit、invokeAll、invokeAny 等方法。至于 ScheduledExecutorService 接口,则是声明了一些和定时任务相关的方法,比如 schedule和scheduleAtFixedRate。线程池的核心实现是在 ThreadPoolExecutor 类中,我们使用 Executors 调用newFixedThreadPool、newSingleThreadExecutor和newCachedThreadPool等方法创建线程池均是 ThreadPoolExecutor 类型。
3. 核心参数分析
核心参数的配置在我的这篇博客《线程池不允许使用Executors去创建,而是通过ThreadPoolExecutor的方式》中有详细介绍,也可以移步参考。
3.1 核心参数简介
线程池的核心实现即 ThreadPoolExecutor 类。该类包含了几个核心属性,这些属性在可在构造方法进行初始化。在介绍核心属性前,我们先来看看 ThreadPoolExecutor 的构造方法,如下:
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
线程池的核心参数有如上七个,各个参数意义如下:
3.2 线程创建规则
在 Java 线程池实现中,线程池所能创建的线程数量受限于 corePoolSize 和 maximumPoolSize 两个参数值。线程的创建时机则和 corePoolSize 以及 workQueue 两个参数有关。
执行图示如下:
3.3 资源回收
考虑到系统资源是有限的,对于线程池超出 corePoolSize 数量的空闲线程应进行回收操作。进行此操作存在一个问题,即回收时机。目前的实现方式是当线程空闲时间超过 keepAliveTime 后,进行回收。除了核心线程数之外的线程可以进行回收,核心线程内的空闲线程也可以进行回收。回收的前提是allowCoreThreadTimeOut属性被设置为 true,通过public void allowCoreThreadTimeOut(boolean) 方法可以设置属性值。
3.4 排队策略
当线程数量大于等于 corePoolSize,workQueue 未满时,则缓存新任务。这里要考虑使用什么类型的容器缓存新任务,通过 JDK 文档介绍,我们可知道有3中类型的容器可供使用,分别是同步队列
,有界队列
和无界队列
。对于有优先级的任务,这里还可以增加优先级队列
。以上所介绍的4中类型的队列,对应的实现类如下:
3.5 拒绝策略
当线程数量大于等于 maximumPoolSize,且 workQueue 已满,则使用拒绝策略处理新任务。Java 线程池提供了4中拒绝策略实现类,AbortPolicy 是线程池实现类所使用的策略。
4. 线程池的操作
4.1 线程的创建与复用
下面是实现类中一些比较重要的成员变量:
private final BlockingQueue<Runnable> workQueue; //任务缓存队列,用来存放等待执行的任务 private final ReentrantLock mainLock = new ReentrantLock(); //线程池的主要状态锁,对线程池状态(比如线程池大小、runState等)的改变都要使用这个锁 private final HashSet<Worker> workers = new HashSet<Worker>(); //用来存放工作集 private volatile long keepAliveTime; //线程存活时间 private volatile boolean allowCoreThreadTimeOut; //是否允许为核心线程设置存活时间 private volatile int corePoolSize; //核心池的大小(即线程池中的线程数目大于这个参数时,提交的任务会被放进任务缓存队列) private volatile int maximumPoolSize; //线程池最大能容忍的线程数 private volatile int poolSize; //线程池中当前的线程数 private volatile RejectedExecutionHandler handler; //任务拒绝策略 private volatile ThreadFactory threadFactory; //线程工厂,用来创建线程 private int largestPoolSize; //用来记录线程池中曾经出现过的最大线程数 private long completedTaskCount; //用来记录已经执行完毕的任务个数
在线程池的实现上,线程的创建是通过线程工厂接口ThreadFactory
的实现类来完成的。默认情况下,线程池使用Executors.defaultThreadFactory()
方法返回的线程工厂实现类。当然,我们也可以通过 public void setThreadFactory(ThreadFactory)
方法进行动态修改。
+----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); } }
4.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) { // 添加工作者对象 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()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); // 将 worker 对象添加到 workers 集合中 workers.add(w); int s = workers.size(); // 更新 largestPoolSize 属性 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 开始执行任务 t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
4.3 关闭线程池
我们可以通过shutdown
和shutdownNow
两个方法关闭线程池。两个方法的区别在于,shutdown 会将线程池的状态设置为SHUTDOWN
,同时该方法还会中断空闲线程。shutdownNow 则会将线程池状态设置为STOP
,并尝试中断所有的线程。中断线程使用的是Thread.interrupt
方法,未响应中断方法的任务是无法被中断的。最后,shutdownNow 方法会将未执行的任务全部返回。
public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); //线程池状态设置为 SHUTDOWN advanceRunState(SHUTDOWN); //中断空闲线程 interruptIdleWorkers(); onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } tryTerminate(); } public List<Runnable> shutdownNow() { List<Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); //线程池状态设置为 SHUTDOWN advanceRunState(STOP); //中断所有线程 interruptWorkers(); //未执行任务 用于return tasks = drainQueue(); } finally { mainLock.unlock(); } tryTerminate(); return tasks; }
调用 shutdown 和 shutdownNow 方法关闭线程池后,就不能再向线程池提交新任务了。对于处于关闭状态的线程池,会使用拒绝策略处理新提交的任务。
5. 线程池大小配置
本节来讨论一个比较重要的话题:如何合理配置线程池大小,仅供参考。
一般需要根据任务的类型来配置线程池大小:
- 如果是CPU密集型任务,就需要尽量压榨CPU,参考值可以设为 NCPU+1
- 如果是IO密集型任务,参考值可以设置为2*NCPU
当然,这只是一个参考值,具体的设置还需要根据实际情况进行调整,比如可以先将线程池大小设置为参考值,再观察任务运行情况和系统负载、资源利用率来进行适当调整。
参考:
https://www.cnblogs.com/dolphin0520/p/3932921.html
《Java并发编程的艺术》
标签:Java,mainLock,int,private,任务,线程,原理,null 来源: https://www.cnblogs.com/zjfjava/p/13908426.html