大家好,又见面了,我是你们的朋友全栈君。
1 前言
线程池是并发编程中一个重要的概念和技术。大多数异步或并发执行任务都会用到线程池。 线程池,正如其名,它是有一定数量的线程的池子,它会执行被提交过来的任务,执行完一个任务后不会马上结束,它们会继续等待或执行新的任务。线程池有两个重要的概念一个是任务队列,另一个是工作者线程 。任务队列是存放任务的容器,工作者线程会依次不断地到队列中获取任务并执行。
线程池有这些优点:
-
① 减少系统资源的消耗。它通过对线程的重用,避免不断创建新线程导致的系统开销。任务过多时,通过排队避免创建过多的线程,减少系统资源的消耗与竞争,确保任务有序完成。
-
②提高响应速度。当任务到达时,任务无需等待线程的创建完成,它得利用已有的线程立即执行任务。
-
③提高线程的可控性。线程是稀缺资源,不能无限制地创建,线程池它对线程能统一分配、调度和销毁。
线程池直接继承于抽象类AbstractExecutorService
,AbstractExecutorService
是对ExecutorSerivice
接口的默认实现,而ExecutorService
又扩展了Executor
接口。
2 处理任务的流程
线程池对任务的处理有它自己定义的流程,它对任务的处理流程如下:
①线程池判断核心线程池里的线程是否都在执行任务。如果不是,则创建一个新的工作线程来执行任务(线程池在开始阶段会尽快让池中的线程数达到设定的核心线程数)。如果核心线程池里的线程都在执行任务,则进入下个流程。
②线程池判断工作队列是否已经满。如果工作队列没有满,则将新提交的任务存储在这个工作队列里(尽量先往阻塞队列中放)。如果工作队列满了,则进入下个流程。
③线程池判断线程池的线程是否都处于工作状态。如果没有,则创建一个新的工作线程来执行任务(工作队列放不下了,只有创建新线程来执行任务)。如果已经满了,则交给饱和策略来处理这个任务。
其基本理念是:①在线程池启动阶段,尽快让池中的线程数达到设定的核心线程数,这里主要从能利用已有线程立即执行之后提交的新任务、避免创建线程而等待的角度考虑;② 在核心线程池满了之后,尽可能向阻塞队列中放入任务,这里是从减少资源消耗的角度考虑,毕竟线程是稀缺资源、不能无限制地创建;③在阻塞队列已满的情况下,已经无法再往队列中放入任务了,此时只能创建新的线程去执行任务,虽然创建线程会消耗系统资源,但是总不能不执行提交的任务;④而在最坏的情况下,线程池中的线程数也达到了设定的最大线程数,此时已无法直接执行任务了,只能按照指定的饱和策略来拒绝任务。
3 线程池的配置
ThreadPoolExecutor
有4个构造方法,分别需要若干个参数,我们主要通过构造方法参数去配置线程池。我们从其参数个数最多的构造方法看起,其他的构造方法都是直接调用这个构造方法来实现的。
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); this.acc = System.getSecurityManager() == null ? null : AccessController.getContext(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }
这个构造方法有7个参数:
1) corePoolSize : 核心线程数. 提交一个任务到线程池时,线程池会创建一个线程来执行任务,即使其他空闲的基本线程能够执行新任务也会创建线程,等到需要执行的任务数大于线程池基本大小时就不再创建。另外还可以调用setCorePoolSize(int)
方法来设置核心线程数。
默认情况下,核心线程不会从预告创建,只有有任务时才创建;核心线程不会因空闲而终止。但以下几个API可以改变这种默认方式。
int prestartAllCoreThreads()
方法,线程池会提前创建并启动所有核心线程。
boolean prestartCoreThread()
方法,创建一个核心线程,若所有核心线程已创建则返回false.
allowCoreThreadTimeOut(boolean)
方法,若参数是true,核心线程会因空闲和终止(和其他非核心线程一样,使用keepAliveTime参数作为最大空闲存活时间)。
2) maximumPoolSize: 线程池可创建的最大线程数。如果队列已满,线程池将创建新的线程执行任务直到达到这个最大线程数。若是使用无界阻塞队列,队列永远也不会满,它不会创建新线程,它会一直往队列中放任务,其结果是一些任务长时间等待、难以被执行。另外还可以调用setMaximumPoolSize(int)
方法来设置最大线程数
3) keepAliveTime:空闲线程存活时间,设置此参数的目的是释放多余的线程资源。它表示线程个数大于corePoolSize时,其他额外空闲线程的存活时间。也就是说,一个非核心线程在空闲等待新任务时,会有一个最长等待时间,若等待时间超过了keepAliveTime,这个线程就会被销毁。若是将此参数设为0,那么所有的线程将一直不会被销毁。若任务较多且任务执行时间较短,可适当增大此参数,提高线程的利用率,避免反复创建新线程。 另外调用setKeepAliveTime(long , TimeUnit )
方法也可设置空闲线程存活时间。
4) unit :参数keepAliveTime的时间单位,可以是“DAYS”(天) 、”HOURS“(时)、“MINUTES”(分)、“SECONDS”(秒)、”MILLISECONDS“(毫秒)、”NANOSECONDS“(纳秒)等
5) workQueue:工作队列, 用于保存等待执行的任务的阻塞队列。之前的文章并发编程中的阻塞队列概述有对阻塞队列做过介绍,这里只对进行SynchronousQueue
特别说明。SynchronousQueue
不存储元素的阻塞队列,当尝试排队时,只有正好有空闲线程正在等待接受任务时才会入队成功,否则总是创建新线程执行任务,直到线程数达到maximumPoolSize ,其吞吐量通常要高于LinkedBlockingQueue
。
6) threadFactory :线程工厂,主要用于为创建出来的线程设置优先级、取个有意义的名字、是否守护线程等。另外还可调用setThreadFactory(ThreadFactory)
方法设置线程工厂。ThreadFactory
是一个接口,它的定义是:
public interface ThreadFactory { Thread newThread(Runnable r); }
ThreadPoolExecutor
的默认实现是Executors
工具类的静态内部DefaultThreadFactory
,这个线程工厂主要是创建一个线程,设置一个名称,设置daemon为false,将优先级设为标准默认优先级,线程名称的格式:”pool-线程池编号-thread-线程编号“。
7) handler:拒绝策略. 当线程池和队列都满了时,表示线程池已经饱和,此时应采取一些特殊的手段来处理这个新任务。反过来说,拒绝策略只有在队列有界且maximumPoolSize有限大时才会被触发。若队列无界,任务一直往队列中放置,任务一直处于排队中,难以得到执行。若队列有有界、maximumPoolSize无限大,则会创建大量的线程,占满CPU和内存,可能导致程序或系统崩溃。
默认情况下线程池会使用AbortPolicy策略,此策略会直接抛出异常。线程池内置有4种拒绝策略,这4种拒绝策略都是ThreadPoolExecutor
的静态内部类。
-
CallerRunsPolicy
, 使用任务提交者的所在线程执行任务; -
AbortPolicy
,直接抛出异常,这是默认的拒绝策略; -
DiscardPolicy
, 不执行任务,将任务丢弃; -
DiscardOldestPolicy
,丢弃队列中最近的任务,然后执行当前任务。
以上4个类都实现了RejectedExecutionHandler
接口,当线程无法接受新任务时,调用拒绝策略的rejectedExecution
方法进行相应处理。
public interface RejectedExecutionHandler { void rejectedExecution(Runnable r, ThreadPoolExecutor executor); }
拒绝策略除了在构造方法中指定外,还可调用线程池的setRejectedExecutionHandler
方法进行设置。
4 提交任务
1)线程池有两组提交单任务的方法
execute(Runnable)
用于提交不需要结果的任务,因此无法确定任务是否完成。 而submit系列方法, submit(Runnable, T)
、submit(Callable<T>)
、submit(Runnable)
都用于提交需要结果的任务。线程池会返回一个Future类型的对象,通过这个Future对象可以判断任务是否执行成功,并且可以通过future的get()方法来获取返回值,get()
方法会阻塞当前线程直到任务完成(使用“submit(Runnable)
”提交任务,get()
方法最终返回null),而使用get(long,TimeUnit)
方法则会阻塞当前线程一段时间后立即返回,此时返回可能任务未完成。
2)线程池有两组批量提交任务组的方法 invokeAll(Collection)
方法用于批量提交任务、等待所有任务完成成后,返回Future的List集合 。invokeAll(Collection, long, TimeUnit)
方法是超时版本的 invokeAll(Collection)
,需要指定超时时间,若超时后还有任务还未完成,这些未完成的任务就会被取消。
invokeAny(Collection)
也用于批量提交任务,但只要有一个任务正常完成(没抛出异常)后,它就返回此任务的结果;在正常返回或异常抛出返回后,其他任务则会被取消(最多只有一个任务能正常执行完成)。invokeAny(Collection, long , TimeUnit )
是超时版本的invokeAny(Collection)
,它对任务的执行耗时做了限制,如果在限定时间内有一任务正常(没抛出异常)完成,就返回此任务的结果 ,其他将任务会被取消;如果没有任务能在限时内成功完成返回,就抛出 TimeoutException; 没有任务正常成功返回(可能是因发生某种异常而返回),将抛出ExecutionException 。
5 关闭线程池
shutdown()
和shutdownNow()
方法都能关闭线程池,它们的处理逻辑是:遍历线程池中的工作者线程,然后逐个调用线程的interrupt方法来中断线程,若某些任务不能响应中断,那么它们就无法终止。但两者在细节上有一些区别,shutdownNow()
首先将线程池的状态设置成STOP,然后尝试停止所有的正在执行或暂停任务的线程,并返回等待执行任务的列表,而shutdown()
只是将线程池的状态设置成SHUTDOWN状态,然后中断所有没有正在执行任务的线程。
只要调用了这两个关闭方法中的任意一个,isShutdown()
方法就会返回true。当所有的任务都已关闭后,才表示线程池关闭成功,这时调用isTerminated()
方法会返回true。至于应该调用哪一种方法来关闭线程池,应该由提交到线程池的任务特性决定,通常调用shutdown()
方法来关闭线程池,如果任务不一定要执行完,则可以调用shutdownNow()
方法。
6 合理配置线程池
要想合理地配置线程池,就必须首先分析任务特性,可以从以下几个角度来分析。
-
任务的性质:CPU密集型任务、IO密集型任务和混合型任务。
-
任务的优先级:高、中和低。
-
任务的执行时间:长、中和短。
-
任务的依赖性:是否依赖其他系统资源,如数据库连接
性质不同的任务可以用不同规模的线程池分开处理。CPU密集型任务应配置尽可能小的线程,如配置Ncpu +1个线程的线程池。由于IO密集型任务线程并不是一直在执行任务,则应配置尽可能多的线程,如2*Ncpu 。混合型的任务,如果可以拆分,将其拆分成一个CPU密集型任务和一个IO密集型任务,只要这两个任务执行的时间相差不是太大,那么分解后执行的吞吐量将高于串行执行的吞吐量。如果这两个任务执行时间相差太大,则没必要进行分解。可以通过Runtime.getRuntime().availableProcessors()
方法获得当前设备的CPU个数。
优先级不同的任务可以使用优先级队列PriorityBlockingQueue
来处理。它可以让优先级高的任务先执行(如果一直有优先级高的任务提交到队列里,那么优先级低的任务可能永远不能执行)
执行时间不同的任务可以交给不同规模的线程池来处理,或者可以使用优先级队列,让执行时间短的任务先执行。依赖数据库连接池的任务,因为线程提交SQL后需要等待数据库返回结果,等待的时间越长,则CPU空闲时间就越长,那么线程数应该设置得越大,这样才能更好地利用CPU
尽可能使用有界队列 。有界队列能增加系统的稳定性和预警能力,如果当时我们设置成无界队列,队列永不可能满,那么线程池的队列就会越来越多,有可能会导致内存溢出、程序崩溃。
7 状态监控
为了监控线程池,我们可以使用一些方法获取线程池的状态信息。
getTaskCount()
: 计划要执行的任务总数
getCompletedTaskCount()
: 线程池已完成的任务数量
getLargestPoolSize()
: 线程池里曾经创建过的最大线程数量。通过这个数据可以知道线程池是否曾经满过。如该数值等于线程池的最大大小,则表示线程池曾经满过。
getActiveCount()
:当前活动(正在执行任务)的工作线程数
getPoolSize()
:当前线程池中的工作线程总数
除此之外,线程池还提供了3个空方法,beforeExecute
方法在执行一个任务前被调用,afterExecute
方法在一个任务完成后被调用,terminated()
方法在线程池停止时被调用。
我们可继承ThreadPoolExecutor
来实现自己的线程池,并以此为基础重写这3个方法来实现自己的监控逻辑。
protected void beforeExecute(Thread t, Runnable r) { } protected void afterExecute(Runnable r, Throwable t) { } protected void terminated() { }
afterExecute
方法注释上写了一个这样的使用示例,它能打印导致任务非正常完成的异常信息。
class ExtendedExecutor extends ThreadPoolExecutor { // ... protected void afterExecute(Runnable r, Throwable t) { super.afterExecute(r, t); if (t == null && r instanceof Future<?>) { try { Object result = ((Future<?>) r).get(); } catch (CancellationException ce) { t = ce; } catch (ExecutionException ee) { t = ee.getCause(); } catch (InterruptedException ie) { //在捕获中断异常后,中断标志将设为false ,这里调用interrupt恢复中断状态 Thread.currentThread().interrupt(); // ignore/reset } } if (t != null) System.out.println(t); } }
参考:《Java并发编程的艺术》、《Java的逻辑》
发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/154359.html原文链接:https://javaforall.cn
【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛
【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...