java线程池ThreadPoolExecutor详解

2024-01-08 02:01:05

一、前言

? ? ? ? 线程池的使用是为了提高资源的使用效率,简单来说就是CPU的使用效率。Java由于天生的多线程特性,通过池化技术可以更好的利用系统资源。

  • 降低资源消耗(线程无限制地创建,然后使用完毕后销毁)
  • 提高响应速度(无须创建线程)
  • 提高线程的可管理性

创建一个新的线程可以通过继承Thread类或者实现Runnable接口来实现,这两种方式创建的线程在运行结束后会被虚拟机销毁,进行垃圾回收,如果线程数量过多,频繁的创建和销毁线程会浪费资源,降低效率。而线程池的引入就很好解决了上述问题,线程池可以更好的创建、维护、管理线程的生命周期,做到复用,提高资源的使用效率,也避免了开发人员滥用new关键字创建线程的不规范行为。

说明:阿里开发手册中明确指出,在实际生产中,线程资源必须通过线程池提供,不允许在应用中显式的创建线程。如果不使用线程池,有可能造成系统创建大量同类线程而导致消耗完内存或者“过度切换”的问题。

接下来主要对Java中线程池核心实现类ThreadPoolExecutor核心参数及工作原理、Executors工具类等,进行说明。

二、ThreadPoolExecutor核心参数

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个参数,分别是corePoolSize、maximumPoolSize、keepAliveTime、unit、workQueue、threadFactory、handler。

1. corePoolSize 线程池核心线程大小
当向线程池提交一个任务时,若线程池已创建的线程数小于corePoolSize,即便此时存在空闲线程,也会通过创建一个新线程来执行该任务,直到已创建的线程数大于或等于corePoolSize。

2. maximumPoolSize 线程池最大线程数量
线程池所允许的最大线程个数。当队列满了,且已创建的线程数小于maximumPoolSize,则线程池会创建新的线程来执行任务。

3.keepAliveTime 空闲线程存活时间
当线程池中线程数大于核心线程数时,线程的空闲时间如果超过线程存活时间,那么这个线程就会被销毁,直到线程池中的线程数小于等于核心线程数。一个线程如果处于空闲状态,并且当前的线程数量大于corePoolSize,那么在指定时间后,这个空闲线程会被销毁,这里的指定时间由keepAliveTime来设定

4.unit 空闲线程存活时间单位
keepAliveTime的计量单位

5.workQueue 工作队列
用于传输和保存等待执行任务的阻塞队,dk中提供了四种工作队列:

  • ArrayBlockingQueue:基于数组的有界阻塞队列,按FIFO排序。新任务进来后,会放到该队列的队尾,有界的数组可以防止资源耗尽问题。当线程池中线程数量达到corePoolSize后,再有新任务进来,则会将任务放入该队列的队尾,等待被调度。如果队列已经是满的,则创建一个新线程,如果线程数量已经达到maxPoolSize,则会执行拒绝策略。
  • LinkedBlockingQuene:基于链表的无界阻塞队列(其实最大容量为Interger.MAX),按照FIFO排序。由于该队列的近似无界性,当线程池中线程数量达到corePoolSize后,再有新任务进来,会一直存入该队列,而不会去创建新线程直到maxPoolSize,因此使用该工作队列时,参数maxPoolSize其实是不起作用的。newFixedThreadPool线程池使用了这个队列,因为没有没有对队列数量做限制所以可能会有OOM的风险,
  • SynchronousQuene:一个不缓存任务的阻塞队列,生产者放入一个任务必须等到消费者取出这个任务。也就是说新任务进来时,不会缓存,而是直接被调度执行该任务,如果没有可用线程,则创建新线程,如果线程数量达到maxPoolSize,则执行拒绝策略。吞吐量通常要高于LinkedBlockingQuene。newCachedThreadPool线程池使用了这个队列。
  • PriorityBlockingQueue:具有优先级的无界阻塞队列,优先级通过参数Comparator实现。
  • DelayQueue:(延迟队列)是一个任务定时周期的延迟执行的队列。根据指定的执行时间从小到大排序,否则根据插入到队列的先后排序。newScheduledThreadPool线程池使用了这个队列。

6.threadFactory 线程工厂
创建一个新线程时使用的工厂,可以用来设定线程名、是否为daemon线程等等,threadFactory创建的线程也是采用new Thread()方式,threadFactory创建的线程名都具有统一的风格:pool-m-thread-n(m为线程池的编号,n为线程池内的线程编号)

创建工厂的两种方式:Executors.defaultThreadFactory()?、new ThreadFactoryBuilder().setNameFormat("task-service-pool-%d").build()

7.handler 拒绝策略
当线程池和队列都满了,再加入线程会执行此策略。

  • CallerRunsPolicy:该策略下,在调用者线程中直接执行被拒绝任务的run方法,除非线程池已经shutdown,则直接抛弃任务
  • AbortPolicy:该策略下,直接丢弃任务,并抛出RejectedExecutionException异常。
  • DiscardPolicy:该策略下,直接丢弃任务,什么都不做。
  • DiscardOldestPolicy:该策略下,抛弃进入队列最早的那个任务,然后尝试把这次拒绝的任务放入队列

8.线程池流程

  1. 判断核心线程池是否已满,没满则创建一个新的工作线程来执行任务。已满则。
  2. 判断任务队列是否已满,没满则将新提交的任务添加在工作队列,已满则。
  3. 判断整个线程池是否已满,没满则创建一个新的工作线程来执行任务,已满则执行饱和策略。
  4. 判断线程池中当前线程数是否大于核心线程数,如果小于,在创建一个新的线程来执行任务,如果大于则
  5. 判断任务队列是否已满,没满则将新提交的任务添加在工作队列,已满则。
  6. 判断线程池中当前线程数是否大于最大线程数,如果小于,则创建一个新的线程来执行任务,如果大于,则执行饱和策略。

三、Executors 创建线程池工具类(不推荐)

Executors 调用了 ThreadPoolExecutor 来创建线程池的,可以帮助用户根据不用的场景,配置不同的参数。

不推荐使用,因为默认保存任务的队列是无界的,使用原生 new ThreadPoolExecutor() 创建线程池根据自己的需求设置参数来创建线程池比较好。

先说总结:

  • FixedThreadPool和SingleThreadExecutor => 允许的请求队列长度为Integer.MAX_VALUE,可能会堆积大量的请求,从而引起OOM异常。
  • CachedThreadPool和ScheduledThreadPool=> 允许创建的线程数为Integer.MAX_VALUE,可能会创建大量的线程,从而引起OOM异常。

execute()和submit()方法
execute(),执行一个任务,没有返回值。
submit(),提交一个线程任务,有返回值。
submit(Callable task)能获取到它的返回值,通过future.get()获取(阻塞直到任务执行完)。一般使用FutureTask+Callable配合使用(IntentService中有体现)。
submit(Runnable task, T result)能通过传入的载体result间接获得线程的返回值。
submit(Runnable task)则是没有返回值的,就算获取它的返回值也是null。

Future.get方法会使取结果的线程进入阻塞状态,知道线程执行完成之后,唤醒取结果的线程,然后返回结果。

java线程池的调优以及监控
线程池的调优(线程池的合理配置)
先从以下几个角度分析任务的特性:

任务的性质:CPU 密集型任务、IO 密集型任务和混合型任务。
任务的优先级:高、中、低。
任务的执行时间:长、中、短。
任务的依赖性:是否依赖其他系统资源,如数据库连接。
任务性质不同的任务可以用不同规模的线程池分开处理。可以通过 Runtime.getRuntime().availableProcessors() 方法获得当前设备的 CPU 个数。

CPU 密集型任务配置 尽可能小的线程,如配置N^cpu+1个线程的线程池。

IO 密集型任务则由于线程并不是一直在执行任务,则配置尽可能多的线程,如2*N^cpu。

混合型任务 如果可以拆分,则将其拆分成一个 CPU 密集型任务和一个 IO 密集型任务。只要这两个任务执行的时间相差不是太大,那么分解后执行的吞吐率要高于串行执行的吞吐率;如果这两个任务执行时间相差太大,则没必要进行分解。

优先级不同的任务可以使用优先级队列 PriorityBlockingQueue 来处理,它可以让优先级高的任务先得到执行。但是,如果一直有高优先级的任务加入到阻塞队列中,那么低优先级的任务可能永远不能执行。
执行时间不同的任务可以交给不同规模的线程池来处理,或者也可以使用优先级队列,让执行时间短的任务先执行。
依赖数据库连接池的任务,因为线程提交 SQL 后需要等待数据库返回结果,线程数应该设置得较大,这样才能更好的利用 CPU。
建议使用有界队列,有界队列能增加系统的稳定性和预警能力。可以根据需要设大一点,比如几千。使用无界队列,线程池的队列就会越来越大,有可能会撑满内存,导致整个系统不可用。② 线程池的监控
可以通过线程池提供的参数读线程池进行监控,有以下属性可以使用:

taskCount:线程池需要执行的任务数量,包括已经执行完的、未执行的和正在执行的。
completedTaskCount:线程池在运行过程中已完成的任务数量,completedTaskCount <= taskCount。
largestPoolSize:线程池曾经创建过的最大线程数量,通过这个数据可以知道线程池是否满过。如等于线程池的最大大小,则表示线程池曾经满了。
getPoolSize: 线程池的线程数量。如果线程池不销毁的话,池里的线程不会自动销毁,所以线程池的线程数量只增不减。
getActiveCount:获取活动的线程数。
通过继承线程池并重写线程池的 beforeExecute,afterExecute 和 terminated 方法,我们可以在任务执行前,执行后和线程池关闭前干一些事情。
如监控任务的平均执行时间,最大执行时间和最小执行时间等。这几个方法在线程池里是空方法,如:

protected void beforeExecute(Thread t, Runnable r) { }

文章来源:https://blog.csdn.net/qq_37080455/article/details/135350638
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。