Java线程池Executor详解

忘是亡心i 2023-09-27 16:23 143阅读 0赞

### 线程池类图 ###

![009fce65916955d0234474d824a7afc6.png][]

我们最常使用的Executors实现创建线程池使用线程主要是用上述类图中提供的类。在上边的类图中，包含了一个Executor框架，它是一个根据一组执行策略的调用调度执行和控制异步任务的框架，目的是提供一种将任务提交与任务如何运行分离开的机制。它包含了三个executor接口：

*  Executor:运行新任务的简单接口
 *  ExecutorService：扩展了Executor，添加了用来管理执行器生命周期和任务生命周期的方法
 *  ScheduleExcutorService：扩展了ExecutorService，支持Future和定期执行任务

### 线程池的好处 ###

*  **降低资源消耗**\-重用存在的线程，减少对象创建、消亡的开销，性能好
 *  **提高响应速度** \-可有效控制最大并发线程数，提高系统资源利用率，同时可以避免过多资源竞争，避免阻塞。当任务到达时，任务可不用等待线程创建就能立即执行
 *  **提高线程的可管理性**\-提供定时执行、定期执行、单线程、并发数控制等功能。

#### new Thread的弊端 ####

*  每次new Thread 新建对象，性能差
 *  线程缺乏统一管理，可能无限制的新建线程，相互竞争，可能占用过多的系统资源导致死机或者OOM（out of memory 内存溢出），这种问题的原因不是因为单纯的new一个Thread，而是可能因为程序的bug或者设计上的缺陷导致不断new Thread造成的。
 *  缺少更多功能，如更多执行、定期执行、线程中断。

### 线程池核心类-ThreadPoolExecutor ###

参数说明：ThreadPoolExecutor一共有七个参数，这七个参数配合起来，构成了线程池强大的功能。

*  **corePoolSize**：核心线程数量
 *  **maximumPoolSize**：线程最大线程数
 *  **workQueue**：阻塞队列，存储等待执行的任务，很重要，会对线程池运行过程产生重大影响
    
     *  当我们提交一个新的任务到线程池，线程池会根据当前池中正在运行的线程数量来决定该任务的处理方式。处理方式有三种： 1、直接切换（SynchronusQueue） 2、无界队列（LinkedBlockingQueue）能够创建的最大线程数为corePoolSize,这时maximumPoolSize就不会起作用了。当线程池中所有的核心线程都是运行状态的时候，新的任务提交就会放入等待队列中。 3、有界队列（ArrayBlockingQueue）最大maximumPoolSize，能够降低资源消耗，但是这种方式使得线程池对线程调度变的更困难。因为线程池与队列容量都是有限的。所以想让线程池的吞吐率和处理任务达到一个合理的范围，又想使我们的线程调度相对简单，并且还尽可能降低资源的消耗，我们就需要合理的限制这两个数量 分配技巧： \[如果想降低资源的消耗包括降低cpu使用率、操作系统资源的消耗、上下文切换的开销等等，可以设置一个较大的队列容量和较小的线程池容量，这样会降低线程池的吞吐量。如果我们提交的任务经常发生阻塞，我们可以调整maximumPoolSize。如果我们的队列容量较小，我们需要把线程池大小设置的大一些，这样cpu的使用率相对来说会高一些。但是如果线程池的容量设置的过大，提高任务的数量过多的时候，并发量会增加，那么线程之间的调度就是一个需要考虑的问题。这样反而可能会降低处理任务的吞吐量。\]
 *  **keepAliveTime**：线程没有任务执行时最多保持多久时间终止（当线程中的线程数量大于corePoolSize的时候，如果这时没有新的任务提交核心线程外的线程不会立即销毁，而是等待，直到超过keepAliveTime）
 *  **unit**：keepAliveTime的时间单位
 *  **threadFactory**：线程工厂，用来创建线程，有一个默认的工场来创建线程，这样新创建出来的线程有相同的优先级，是非守护线程、设置好了名称）
 *  **rejectHandler**：当拒绝处理任务时(阻塞队列满)的策略（AbortPolicy默认策略直接抛出异常、CallerRunsPolicy用调用者所在的线程执行任务、DiscardOldestPolicy丢弃队列中最靠前的任务并执行当前任务、DiscardPolicy直接丢弃当前任务）

![8e11a9aa126e6494f27f66d88645b022.png][]

> corePoolSize、maximumPoolSize、workQueue 三者关系：如果运行的线程数小于corePoolSize的时候，直接创建新线程来处理任务。即使线程池中的其他线程是空闲的。如果运行中的线程数大于corePoolSize且小于maximumPoolSize时，那么只有当workQueue满的时候才创建新的线程去处理任务。如果corePoolSize与maximumPoolSize是相同的，那么创建的线程池大小是固定的。这时有新任务提交，当workQueue未满时，就把请求放入workQueue中。等待空线程从workQueue取出任务。如果workQueue此时也满了，那么就使用另外的拒绝策略参数去执行拒绝策略。

初始化方法：由七个参数组合成四个初始化方法

![d0ca98ce8883b4872e9cd8296eab6d50.png][]

其他方法：

execute();  //提交任务，交给线程池执行  
    submit();//提交任务，能够返回执行结果 execute+Future
    shutdown();//关闭线程池，等待任务都执行完
    shutdownNow();//关闭线程池，不等待任务执行完
    getTaskCount();//线程池已执行和未执行的任务总数
    getCompleteTaskCount();//已完成的任务数量
    getPoolSize();//线程池当前的线程数量
    getActiveCount();//当前线程池中正在执行任务的线程数量

线程池生命周期：

![d02244f5125b9a5cbe60a152b4383621.png][]

*  running：能接受新提交的任务，也能处理阻塞队列中的任务
 *  shutdown：不能处理新的任务，但是能继续处理阻塞队列中任务
 *  stop：不能接收新的任务，也不处理队列中的任务
 *  tidying：如果所有的任务都已经终止了，这时有效线程数为0
 *  terminated：最终状态

#### 使用Executors创建线程池 ####

使用Executors可以创建四种线程池：分别对应上边提到的四种线程池初始化方法

Executors.newCachedThreadPool

newCachedThreadPool是一个根据需要创建新线程的线程池，当一个任务提交时，corePoolSize为0不创建核心线程，SynchronousQueue是一个不存储元素的队列，可以理解为队里永远是满的，因此最终会创建非核心线程来执行任务。 对于非核心线程空闲60s时将被回收。因为Integer.MAX\_VALUE非常大，可以认为是可以无限创建线程的，在资源有限的情况下容易引起OOM异常。

//创建newCachedThreadPool线程池源码
    public static ExecutorService newCachedThreadPool() {
            /**
            *corePoolSize： 0，核心线程池的数量为0
            *maximumPoolSize：  Integer.MAX_VALUE，可以认为最大线程数是无限的
            *keepAliveTime： 60L
            *unit: 秒
            *workQueue: SynchronousQueue
            **/
            return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                          60L, TimeUnit.SECONDS,
                                          new SynchronousQueue<Runnable>());
        }

使用案例：

public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            final int index = i;
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    log.info("task:{}",index);
                }
            });
        }
    }

值得注意的一点是，newCachedThreadPool的返回值是ExecutorService类型，该类型只包含基础的线程池方法，但却不包含线程监控相关方法，因此在使用返回值为ExecutorService的线程池类型创建新线程时要考虑到具体情况。

![4384dc9c61209a755470872ee6ec7dd3.png][]

Executors.newSingleThreadExecutor

newSingleThreadExecutor是单线程线程池，只有一个核心线程，用唯一的一个共用线程执行任务，保证所有任务按指定顺序执行（FIFO、优先级…）

//newSingleThreadExecutor创建线程池源码
    public static ExecutorService newSingleThreadExecutor() {
        /**
          *  corePoolSize : 1，核心线程池的数量为1
    
          *  maximumPoolSize : 1，只可以创建一个非核心线程
    
          *  keepAliveTime : 0L
    
          *  unit => 秒
    
          *  workQueue => LinkedBlockingQueue
          **/
            return new FinalizableDelegatedExecutorService
                (new ThreadPoolExecutor(1, 1,
                                        0L, TimeUnit.MILLISECONDS,
                                        new LinkedBlockingQueue<Runnable>()));
        }

当一个任务提交时，首先会创建一个核心线程来执行任务，如果超过核心线程的数量，将会放入队列中，因为LinkedBlockingQueue是长度为Integer.MAX\_VALUE的队列，可以认为是无界队列，因此往队列中可以插入无限多的任务，在资源有限的时候容易引起OOM异常，同时因为无界队列，maximumPoolSize和keepAliveTime参数将无效，压根就不会创建非核心线程。

Executors.newFixedThreadPool

定长线程池，核心线程数和最大线程数由用户传入，可以设置线程的最大并发数，超出在队列等待

//newFixedThreadPool创建线程池源码
    public static ExecutorService newFixedThreadPool(int nThreads) {
            /**
              *  corePoolSize : 核心线程的数量为自定义输入nThreads
    
              *  maximumPoolSize ： 最大线程的数量为自定义输入nThreads
    
              *  keepAliveTime ： 0L
    
              *  unit ： 秒
    
              *  workQueue ： LinkedBlockingQueue
              **/
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>());
        }

newFixedThreadPool和SingleThreadExecutor类似，唯一的区别就是核心线程数不同，并且由于使用的是LinkedBlockingQueue，在资源有限的时候容易引起OOM异常。

Executors.newScheduledThreadPool

定长线程池，核心线程数由用户传入，支持定时和周期任务执行

//newScheduledThreadPool创建线程池源码
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
            return new ScheduledThreadPoolExecutor(corePoolSize);
    }
    
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        /**
          *  corePoolSize : 核心线程的数量为自定义输入corePoolSize
    
          *  maximumPoolSize ： 最大线程的数量为Integer.MAX_VALUE
    
          *  keepAliveTime ： 0L
    
          *  unit ： 纳秒
    
          *  workQueue ： DelayedWorkQueue
          **/
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

当一个任务提交时，corePoolSize为自定义输入，首先创建核心线程，核心线程满了之后，因此最终会创建非核心线程来执行任务。非核心线程使用后将被回收。因为Integer.MAX\_VALUE非常大，可以认为是可以无限创建线程的，在资源有限的情况下容易引起OOM异常。因为使用的DelayedWorkQueue可以实现定时和周期任务。 ScheduledExecutorService提供了三种方法可以使用：

![392aa21e311308cf2c0b1891d0d58d3f.png][]

schedule：延迟后执行任务 scheduleAtFixedRate：以指定的速率执行任务 scheduleWithFixedDelay：以指定的延迟执行任务 使用案例：

public static void main(String[] args) {
    
            ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
    
    //        executorService.schedule(new Runnable() {
    //            @Override
    //            public void run() {
    //                log.warn("schedule run");
    //            }
    //         //延迟3秒后执行
    //        }, 3, TimeUnit.SECONDS);
            //        executorService.shutdown();
    
    //        executorService.scheduleWithFixedDelay(new Runnable() {
    //            @Override
    //            public void run() {
    //                log.warn("scheduleWithFixedDelay run");
    //            }
    //            //延迟一秒后每隔3秒执行
    //        }, 1, 3, TimeUnit.SECONDS);
    
            executorService.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    log.warn("schedule run");
                }
                //延迟一秒后每隔3秒执行
            }, 1, 3, TimeUnit.SECONDS);
    
            /**
             * 定时器调度，不推荐使用，推荐ScheduledExecutorService调度
             */
    //        Timer timer = new Timer();
    //        timer.schedule(new TimerTask() {
    //            @Override
    //            public void run() {
    //                log.warn("timer run");
    //            }
    //        //从当前时间每隔5秒执行
    //        }, new Date(), 5 * 1000);
        }
    复制代码

总结

*  FixedThreadPool和SingleThreadExecutor 允许的请求队列长度为Integer.MAX\_VALUE，可能会堆积大量的请求，从而引起OOM异常
 *  CachedThreadPool 和newScheduledThreadPool允许创建的线程数为Integer.MAX\_VALUE，可能会创建大量的线程，从而引起OOM异常

这就是为什么禁止使用Executors去创建线程池，而是推荐自己去创建ThreadPoolExecutor的原因

### 如何定义线程池参数 ###

**CPU密集型 ：** 线程池的大小推荐为CPU数量 + 1，CPU数量可以根据Runtime.availableProcessors方法获取 **IO密集型 ：** CPU数量 \* CPU利用率 \* (1 + 线程等待时间/线程CPU时间) **混合型 ：** 将任务分为CPU密集型和IO密集型，然后分别使用不同的线程池去处理，从而使每个线程池可以根据各自的工作负载来调整 **阻塞队列 ：** 推荐使用有界队列，有界队列有助于避免资源耗尽的情况发生 **拒绝策略 ：** 默认采用的是AbortPolicy拒绝策略，直接在程序中抛出RejectedExecutionException异常【因为是运行时异常，不强制catch】，这种处理方式不够优雅。处理拒绝策略有以下几种比较推荐：

*  在程序中捕获RejectedExecutionException异常，在捕获异常中对任务进行处理。针对默认拒绝策略
 *  使用CallerRunsPolicy拒绝策略，该策略会将任务交给调用execute的线程执行【一般为主线程】，此时主线程将在一段时间内不能提交任何任务，从而使工作线程处理正在执行的任务。此时提交的线程将被保存在TCP队列中，TCP队列满将会影响客户端，这是一种平缓的性能降低
 *  自定义拒绝策略，只需要实现RejectedExecutionHandler接口即可
 *  如果任务不是特别重要，使用DiscardPolicy和DiscardOldestPolicy拒绝策略将任务丢弃也是可以的

如果使用Executors的静态方法创建ThreadPoolExecutor对象，可以通过使用Semaphore对任务的执行进行限流也可以避免出现OOM异常

[009fce65916955d0234474d824a7afc6.png]: https://img-blog.csdnimg.cn/img_convert/009fce65916955d0234474d824a7afc6.png
[8e11a9aa126e6494f27f66d88645b022.png]: https://img-blog.csdnimg.cn/img_convert/8e11a9aa126e6494f27f66d88645b022.png
[d0ca98ce8883b4872e9cd8296eab6d50.png]: https://img-blog.csdnimg.cn/img_convert/d0ca98ce8883b4872e9cd8296eab6d50.png
[d02244f5125b9a5cbe60a152b4383621.png]: https://img-blog.csdnimg.cn/img_convert/d02244f5125b9a5cbe60a152b4383621.png
[4384dc9c61209a755470872ee6ec7dd3.png]: https://img-blog.csdnimg.cn/img_convert/4384dc9c61209a755470872ee6ec7dd3.png
[392aa21e311308cf2c0b1891d0d58d3f.png]: https://img-blog.csdnimg.cn/img_convert/392aa21e311308cf2c0b1891d0d58d3f.png