多线程(三) | 彻底搞懂线程池-ThreadPoolExecutor

2022-05-10 17:17:18 浏览数 (3)

上篇文章我们讲解了Executors这个线程池工厂类的用法,介绍了它所提供的一些常用线程池工具的用法,今天我们来继续深入学习。

一、ThreadPoolExecutor

上面我们着重介绍了通过Executors 工具类获得的各类连接池的使用。上面的几个连接池中,除了newWorkStealingPool 这个连接池,剩下的连接池,其实底层都是使用ThreadPoolExecutor 来创建的。包括newScheduledThreadPool() 底层是通过ScheduledThreadPoolExecutor 类构建的,而ScheduledThreadPoolExecutor 类本身也是ThreadPoolExecutor的子类。

所以ThreadPoolExecutor 类是线程池中非常重要的一个类,也是面试过程中的一个高频考点。今天我们来重难点聊聊。

1.1 Executors类中线程池解读

首先我们先看一下上面的几个连接池的底层实现方式:

newSingleThreadExecutor

代码语言:javascript复制
public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

代码中我们可以看到,这个单线程的线程池,创建了一个ThreadPoolExecutor 对象,然后又将其包装成了FinalizableDelegatedExecutorService 对象。

这里我们重点要观察创建ThreadPoolExecutor 的构造参数:先列出来,后面比较。

代码语言:javascript复制
new ThreadPoolExecutor(1, 1,
                            0L, TimeUnit.MILLISECONDS,
                            new LinkedBlockingQueue<Runnable>())

newFixedThreadPool

代码语言:javascript复制
public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

newCachedThreadPool

代码语言:javascript复制
public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

newScheduledThreadPool

代码语言:javascript复制
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

// ScheduledThreadPoolExecutor
public ScheduledThreadPoolExecutor(int corePoolSize) {
    // super就是调用ThreadPoolExecutor 的构造方法
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

通过上面的代码解读,我们发现,创建线程池基本都是直接或间接调用ThreadPoolExecutor 的构造方法实现的,那么我们很有必要了解一下ThreadPoolExecutor这个类。

1.2 ThreadPoolExecutor类

ThreadPoolExecutor 这个类就是Java中提供给我们用于创建线程池的类。而在《阿里巴巴-Java开发手册》中也明确提到了,不要使用Executor创建线程池,而是通过ThreadPoolExecutor的方式创建,这样的处理方式能让编写代码的人更加明确线程池的运行规则,规避资源耗尽的风险。所以如果我们需要使用线程池,尽量通过ThreadPoolExecutor 去创建,那么我们有必要了解一下ThreadPoolExecutor类创建线程池的方法:

ThreadPoolExecutor 类中共有四个构造方法

这种多个构造方法的,一般参数最多的就是最核心的,因为内部基本都是调用的参数最多的,只不过有一些传入了一些默认参数而已。

那么我们就来看一下参数最多的构造方法:

代码语言:javascript复制
    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
    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;
    }

参数的含义如下:

  • corePoolSize: 指定了线程池中的线程数量,它的数量决定了添加的任务是开辟新的线程去执行,还是放到workQueue任务队列中去;
  • maximumPoolSize: 指定了线程池中的最大线程数量,这个参数会根据你使用的workQueue任务队列的类型,决定线程池会开辟的最大线程数量
  • keepAliveTime: 当线程池中空闲线程数量超过corePoolSize时,多余的线程会在多长时间内被销毁;
  • unit: keepAliveTime的单位
  • workQueue: 任务队列,被添加到线程池中,但尚未被执行的任务;它一般分为直接提交队列、有界任务队列、无界任务队列、优先任务队列几种;
  • threadFactory: 线程工厂,用于创建线程,一般用默认即可;
  • handler: 拒绝策略;当任务太多来不及处理时,如何拒绝任务;

线程池执行线程任务的主要流程是:

一个任务通过 execute(Runnable)方法被添加到线程池,任务就是一个 Runnable类型的对象,任务的执行方法就是Runnable类型对象的run()方法。当一个任务通过execute(Runnable)方法欲添加到线程池时:

  • 如果此时线程池中的数量小于corePoolSize,即使线程池中的线程都处于空闲状态,也要创建新的线程来处理被添加的任务。
  • 如果此时线程池中的数量等于 corePoolSize,但是缓冲队列 workQueue未满,那么任务被放入缓冲队列。
  • 如果此时线程池中的数量大于corePoolSize,缓冲队列workQueue满,并且线程池中的数量小于maximumPoolSize,建新的线程来处理被添加的任务。
  • 如果此时线程池中的数量大于corePoolSize,缓冲队列workQueue满,并且线程池中的数量等于maximumPoolSize,那么通过 handler所指定的策略来处理此任务。也就是:处理任务的优先级为:核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize,如果三者都满了,使用handler处理被拒绝的任务。
  • 当线程池中的线程数量大于 corePoolSize时,如果某线程空闲时间超过keepAliveTime,线程将被终止。这样,线程池可以动态的调整池中的线程数。

1.3 线程池中的重要参数-阻塞队列

接下我们介绍一下 线程池中的比较重要的参数: workQueue,这个队列是一个BlockingQueue workQueue, 也就是一个 Runnable类型的阻塞队列,Runnable类型很好理解,就是我们等待执行的任务么,放到这个队列中。那么什么是BlockingQueue呢:

BlockingQueue:即阻塞队列,什么是阻塞队列呢,就是在某些情况下对阻塞队列的访问可能会造成阻塞。主要有两种情况:

  • 当队列满了的时候进行入队操作;
  • 当队列空了进行出队列操作

那么线程池中为什么要使用阻塞队列呢?我们就以取数据为例,使用阻塞队列可以保证如果队列为空的时候,在读取数据时这个方法是阻塞的,当我们此时又来了一个任务,有可以保证新来的任务能够被获取出来。如果不适用阻塞对列,我们就需要设计一个新加入数据时同时线程来拿数据,这个就比较麻烦了,而使用阻塞队列就可以解决这个问题。

同理当队列满的时候,添加元素是阻塞的,直到有队列中有位置,这样就可以保证的在合理利用资源的前提下,新加入的数据不丢失。

在线程池中常用的阻塞队列实现有如下几个:

  • 直接提交队列-SynchronousQueue SynchronousQueue是一个特殊的BlockingQueue,它没有容量,没执行一个插入操作就会阻塞,需要再执行一个删除操作才会被唤醒,反之每一个删除操作也都要等待对应的插入操作。 使用SynchronousQueue队列,提交的任务不会被保存,总是会马上提交执行。如果用于执行任务的线程数量小于maximumPoolSize,则尝试创建新的进程,如果达到maximumPoolSize设置的最大值,则根据你设置的handler执行拒绝策略。因此这种方式你提交的任务不会被缓存起来,而是会被马上执行,在这种情况下,你需要对你程序的并发量有个准确的评估,才能设置合适的maximumPoolSize数量,否则很容易就会执行拒绝策略; newCachedThreadExecutor 就是使用的直接提交队列
  • 有界的任务队列-ArrayBlockingQueue 使用ArrayBlockingQueue有界任务队列,若有新的任务需要执行时,线程池会创建新的线程,直到创建的线程数量达到corePoolSize时,则会将新的任务加入到等待队列中。若等待队列已满,即超过ArrayBlockingQueue初始化的容量,则继续创建线程,直到线程数量达到maximumPoolSize设置的最大线程数量,若大于maximumPoolSize,则执行拒绝策略。在这种情况下,线程数量的上限与有界任务队列的状态有直接关系,如果有界队列初始容量较大或者没有达到超负荷的状态,线程数将一直维持在corePoolSize以下,反之当任务队列已满时,则会以maximumPoolSize为最大线程数上限。
  • 无界的任务队列-LinkedBlockingQueue 使用无界任务队列,线程池的任务队列可以无限制的添加新的任务,而线程池创建的最大线程数量就是你corePoolSize设置的数量,也就是说在这种情况下maximumPoolSize这个参数是无效的,哪怕你的任务队列中缓存了很多未执行的任务,当线程池的线程数达到corePoolSize后,就不会再增加了;若后续有新的任务加入,则直接进入队列等待,当使用这种任务队列模式时,一定要注意你任务提交与处理之间的协调与控制,不然会出现队列中的任务由于无法及时处理导致一直增长,直到最后资源耗尽的问题 我们前面提到的newSingleThreadExecutornewFixedThreadPool使用的就是无界队列

大家在选择的时候,还是要根据具体的使用场景,要考虑到线程执行的任务的时间,数据可丢失的忍受程度,内存的大小等进行合理的选择,有时候选择不慎,就会导致线程池的使用出现一些系统级别的问题。

  • FixedThreadPool 和 SingleThreadPool : 允许的请求队列长度为 Integer.MAX_VALUE ,可能会堆积大量的请求,从而导致 OOM 。
  • CachedThreadPool 和 ScheduledThreadPool : 允许的创建线程数量为 Integer.MAX_VALUE ,可能会创建大量的线程,从而导致 OOM

1.4 拒绝策略

什么是拒绝策略呢?

JDK主要提供了4种饱和策略供选择。4种策略都做为静态内部类在ThreadPoolExcutor中进行实现。

拒绝策略提供顶级接口 RejectedExecutionHandler ,其中方法 rejectedExecution 即定制具体的拒绝策略的执行逻辑。

jdk默认提供了四种拒绝策略:

  • CallerRunsPolicy - 当触发拒绝策略,只要线程池没有关闭的话,则使用调用线程直接运行任务。一般并发比较小,性能要求不高,不允许失败。但是,由于调用者自己运行任务,如果任务提交速度过快,可能导致程序阻塞,性能效率上必然的损失较大
  • AbortPolicy - 丢弃任务,并抛出拒绝执行 RejectedExecutionException 异常信息。线程池默认的拒绝策略。必须处理好抛出的异常,否则会打断当前的执行流程,影响后续的任务执行。
  • DiscardPolicy - 直接丢弃,其他啥都没有
  • DiscardOldestPolicy - 当触发拒绝策略,只要线程池没有关闭的话,丢弃阻塞队列 workQueue 中最老的一个任务,并将新任务加入

了解了上面的这些概念,我们反过来再看看我们之前的线程池工具:我们以newFixedThreadPool 为例:

代码语言:javascript复制
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

调用ThreadPoolExecutor 的构造方法, 将核心线程数和最大线程数都设置为我们的参数,所以也就代表了这个线程池中的线程数量最大就是 我们传入的参数。同时持续时间为0秒,代表线程如果没有任务了就会自动销毁。

该线程池使用了LinkedBlockingQueue 阻塞队列,这是属于无界的阻塞队列,也就是当我的线程都被占用,如果还有任务不断放入队列中,当任务堆积到一定程度,就到导致java内存溢出。

这里没有显示的传递拒绝策略参数。我们到构造方法中继续查看:

发现如果不传的话使用的是默认的拒绝策略:

而默认的拒绝策略就是: AbortPolicy: 丢弃任务,并抛出拒绝执行 RejectedExecutionException 异常信息。

剩下几个大家也可以尝试去读一下代码。

二、线程池源码浅析

当我们创建了不同类型的线程池,本质上就是ThreadPoolExecutor 中的参数值不同,比如核心线程数,最大线程数,拒绝策略和阻塞队列等。

当我们提交一个线程任务的时候,执行的是线程池中的 submit方法:

submit方法回去调用线程池中和核心方法: execute()

在execute方法中会根据我们的核心线程数好队列中的任务获取情况来判断,该任务是接收还是拒绝,拒绝的话就会直接调用我们拒绝策略中的rejectedExecution 方法。 接收的话,就会调用 addWorder方法。这里面使用原子量 clt用来做一致性校验。

在addWorker中,会使用锁,状态,CAS进行判断,并把Runnable封装成Worker类型,创建线程来执行相关任务。

0 人点赞