从0到1玩转线程池

兜里有辣条
3,034 阅读19分钟

我们一般不会选择直接使用线程类Thread进行多线程编程，而是使用更方便的线程池来进行任务的调度和管理。线程池就像共享单车，我们只要在我们有需要的时候去获取就可以了。甚至可以说线程池更棒，我们只需要把任务提交给它，它就会在合适的时候运行了。但是如果直接使用Thread类，我们就需要在每次执行任务时自己创建、运行、等待线程了，而且很难对线程进行整体的管理，这可不是一件轻松的事情。既然我们已经有了线程池，那还是把这些麻烦事交给线程池来处理吧。

之前一篇介绍线程池使用及其源码的文章篇幅太长了、跨度太大了一些，感觉不是很好理解。所以我把内容重新组织了一下，拆为了两篇文章，并且补充了一些内容，希望能让大家更容易地理解相关内容。

这篇文章将从线程池的概念与一般使用入手，首先介绍线程池的一般使用。然后详细介绍线程池中常用的可配置项，例如任务队列、拒绝策略等，最后会介绍四种常用的线程池配置。通过这篇文章，大家可以熟练掌握线程池的使用方式，在实践中游刃有余地使用线程池对线程进行灵活的调度。

阅读本文需要对多线程编程有基本的认识，例如什么是线程、多线程解决的是什么问题等。不了解的读者可以参考一下我之前发布的一篇文章《这一次，让我们完全掌握Java多线程(2/10)》

一般我们最常用的线程池实现类是ThreadPoolExecutor，我们接下来会介绍这个类的基本使用方法。JDK已经对线程池做了比较好的封装，相信这个过程会非常轻松。

线程池的基本使用

创建线程池

既然线程池是一个Java类，那么最直接的使用方法一定是new一个ThreadPoolExecutor类的对象，例如ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>() )。那么这个构造器的里每个参数是什么意思呢？我们可以暂时不用关心这些细节，继续完成线程池的使用之旅，稍后再回头来研究这个问题。

提交任务

当创建了一个线程池之后我们就可以将任务提交到线程池中执行了。提交任务到线程池中相当简单，我们只要把原来传入Thread类构造器的Runnable对象传入线程池的execute方法或者submit方法就可以了。execute方法和submit方法基本没有区别，两者的区别只是submit方法会返回一个Future对象，用于检查异步任务的执行情况和获取执行结果（异步任务完成后）。

我们可以先试试如何使用比较简单的execute方法，代码例子如下：

public class ThreadPoolTest {

    private static int count = 0;

    public static void main(String[] args) throws Exception {
        Runnable task = new Runnable() {
            public void run() {
                for (int i = 0; i < 1000000; ++i) {
                    synchronized (ThreadPoolTest.class) {
                        count += 1;
                    }
                }
            }
        };

        // 重要：创建线程池
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 1, 0L,
        TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

        // 重要：向线程池提交两个任务
        threadPool.execute(task);
        threadPool.execute(task);

        // 等待线程池中的所有任务完成
        threadPool.shutdown();
        while (!threadPool.awaitTermination(1L, TimeUnit.MINUTES)) {
            System.out.println("Not yet. Still waiting for termination");
        }
        
        System.out.println("count = " + count);
    }
}

运行之后得到的结果是两百万，我们成功实现了第一个使用线程池的程序。那么回到刚才的问题，创建线程池时传入的那些参数有什么作用的呢？

深入解析线程池

创建线程池的参数

下面是ThreadPoolExecutor的构造器定义：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

各个参数分别表示下面的含义：

  1. corePoolSize，核心线程池大小，一般线程池会至少保持这么多的线程数量；
  2. maximumPoolSize，最大线程池大小，也就是线程池最大的线程数量；
  3. keepAliveTime和unit共同组成了一个超时时间，keepAliveTime是时间数量，unit是时间单位，单位加数量组成了最终的超时时间。这个超时时间表示如果线程池中包含了超过corePoolSize数量的线程，则在有线程空闲的时间超过了超时时间时该线程就会被销毁；
  4. workQueue是任务的阻塞队列，在没有线程池中没有足够的线程可用的情况下会将任务先放入到这个阻塞队列中等待执行。这里传入的队列类型就决定了线程池在处理这些任务时的策略，具体类型会在下文中介绍；
  5. threadFactory，线程的工厂对象，线程池通过该对象创建线程。我们可以通过传入自定义的实现了ThreadFactory接口的类来修改线程的创建逻辑，可以不传，默认使用Executors.defaultThreadFactory()作为默认的线程工厂；
  6. handler，拒绝策略，在线程池无法执行或保存新提交的任务时进行处理的对象，常用的有以下几种策略类：
    • ThreadPoolExecutor.AbortPolicy，默认策略，行为是直接抛出RejectedExecutionException异常
    • ThreadPoolExecutor.CallerRunsPolicy，用调用者所在的线程来执行任务
    • ThreadPoolExecutor.DiscardOldestPolicy，丢弃阻塞队列中最早提交的任务，并重试execute方法
    • ThreadPoolExecutor.DiscardPolicy，静默地直接丢弃任务，不返回任何错误

看到这里可能大部分读者并不能理解每个参数具体的作用，接下来我们就通过线程池源代码中使用了这些参数配置的代码来深入理解每一个参数的意义。

execute方法的实现

我们一般会使用execute方法提交我们的任务，那么线程池在这个过程中做了什么呢？在ThreadPoolExecutor类的execute()方法的源代码中，我们主要做了四件事：

  1. 如果当前线程池中的线程数小于核心线程数corePoolSize，则通过threadFactory创建一个新的线程，并把入参中的任务作为第一个任务传入该线程；
  2. 如果当前线程池中的线程数已经达到了核心线程数corePoolSize，那么就会通过阻塞队列workerQueue的offer方法来将任务添加到队列中保存，并等待线程空闲后进行执行；
  3. 如果线程数已经达到了corePoolSize且阻塞队列中无法插入该任务（比如已满），那么线程池就会再增加一个线程来执行该任务，除非线程数已经达到了最大线程数maximumPoolSize；
  4. 如果确实已经达到了最大线程数，那么就会通过拒绝策略对象handler拒绝这个任务。

总体上的执行流程如下，左侧的实心黑点代表流程开始，下方的黑色同心圆代表流程结束：

上面提到了线程池构造器参数中除了超时时间之外的所有参数的作用，相信大家根据上面的流程已经可以理解每个参数的意义了。但是有一个名词我们还一直没有深入讲解，那就是阻塞队列的含义。

线程池中的阻塞队列

线程池中的阻塞队列专门用于存放需要等待线程空闲的待执行任务，而阻塞队列是这样的一种数据结构，它是一个队列（类似于一个List），可以存放0到N个元素。我们可以对这个队列进行插入和弹出元素的操作，弹出操作可以理解为是一个获取并从队列中删除一个元素的操作。当队列中没有元素时，对这个队列的获取操作将会被阻塞，直到有元素被插入时才会被唤醒；当队列已满时，对这个队列的插入操作将会被阻塞，直到有元素被弹出后才会被唤醒。

这样的一种数据结构非常适合于线程池的场景，当一个工作线程没有任务可处理时就会进入阻塞状态，直到有新任务提交后才被唤醒。

在线程池中，不同的阻塞队列类型会被线程池的行为产生不同的影响，下面是三种我们最常用的阻塞队列类型：

  1. 直连队列，以SynchronousQueue类为代表，队列不会存储任何任务。当有任务提交线程试图向队列中添加待执行任务时会被阻塞，直到有任务处理线程试图从队列中获取待执行任务时会与阻塞状态中的任务提交线程发生直接联系，由任务提交线程把任务直接交给任务执行线程；
  2. 无界队列，以LinkedBlockingQueue类为代表，队列中可以存储无限数量的任务。这种队列永远不会因为队列已满导致任务放入队列失败，所以结合前面介绍的流程我们可以发现，当使用无界队列时，线程池中的线程最多只能达到核心线程数就不会再增长了，最大线程数maximumPoolSize参数不会产生作用；
  3. 有界队列，以ArrayBlockingQueue类为代表，可以保存固定数量的任务。这种队列在实践中比较常用，因为它既不会因为保存太多任务导致资源消耗过多（无界队列），又不会因为任务提交线程被阻塞而影响到系统的性能（直连队列）。总体上来说，有界队列在实际效果上比较均衡。

阅读execute方法的源码

在IDE中，例如IDEA里，我们可以点击我们样例代码里的ThreadPoolExecutor类跳转到JDK中ThreadPoolExecutor类的源代码。在源代码中我们可以看到很多java.util.concurrent包的缔造者大牛“Doug Lea”所留下的各种注释，下面的图片就是该类源代码的一个截图。

这些注释的内容非常有参考价值，建议有能力的读者朋友可以自己阅读一遍。下面，我们就一步步地抽丝剥茧，来揭开线程池类ThreadPoolExecutor源代码的神秘面纱。不过这一步并不是必须的，可以跳过。

下面是ThreadPoolExecutor中execute方法带有中文解释的源代码，有兴趣的朋友可以和上面的流程对照起来参考一下：

public void execute(Runnable command) {
    // 检查提交的任务是否为空
    if (command == null)
        throw new NullPointerException();
    
    // 获取控制变量值
    int c = ctl.get();
    // 检查当前线程数是否达到了核心线程数
    if (workerCountOf(c) < corePoolSize) {
        // 未达到核心线程数，则创建新线程
        // 并将传入的任务作为该线程的第一个任务
        if (addWorker(command, true))
            // 添加线程成功则直接返回，否则继续执行
            return;

        // 因为前面调用了耗时操作addWorker方法
        // 所以线程池状态有可能发生了改变，重新获取状态值
        c = ctl.get();
    }

    // 判断线程池当前状态是否是运行中
    // 如果是则调用workQueue.offer方法将任务放入阻塞队列
    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);
    }
    // 如果放入阻塞队列失败（如队列已满），则添加一个线程
    else if (!addWorker(command, false))
        // 如果添加线程失败（如已经达到了最大线程数），则拒绝任务
        reject(command);
}

在这段源代码中，我们可以看到，线程池是通过addWorker方法来创建线程的，这里的这个Worker指的就是ThreadPoolExecutor类中用来对线程进行包装和管理的Worker类对象。如果想了解Worker类的具体执行流程可以阅读一下下一篇深入剖析线程池的任务执行流程的文章。

超时时间

那么还有一个我们没有提到的超时时间在这个过程中发挥了什么作用呢？从前面我们可以看出，线程数量被划分为了核心线程数和最大线程数。当线程没有任务可执行时会阻塞在从队列中获取新任务这个操作上，这时我们称这个线程为空闲线程，一旦有新任务被提交，则该线程就会退出阻塞状态并开始执行这个新任务。

如果当前线程池中的线程总数大于核心线程数，那么只要有线程的空闲时间超过了超时时间，那么这个线程就会被销毁；如果线程池中的线程总数小于等于核心线程数，那么超时线程就不会被销毁了（除了一些特殊情况外）。这也就是超时时间参数所发挥的作用了。

其他线程池操作

关闭线程池

在之前使用线程池执行任务的代码中为了等待线程池中的所有任务执行完已经使用了shutdown()方法，这是关闭线程池的一种方法。对于ThreadPoolExecutor，关闭线程池的方法主要有两个：

  1. shutdown()，有序关闭线程池，调用后线程池会让已经提交的任务完成执行，但是不会再接受新任务。
  2. shutdownNow()，直接关闭线程池，线程池中正在运行的任务会被中断，正在等待执行的任务不会再被执行，但是这些还在阻塞队列中等待的任务会被作为返回值返回。

监控线程池运行状态

我们可以通过调用线程池对象上的一些方法来获取线程池当前的运行信息，常用的方法有：

  • getTaskCount，线程池中已完成、执行中、等待执行的任务总数估计值。因为在统计过程中任务会发生动态变化，所以最后的结果并不是一个准确值；
  • getCompletedTaskCount，线程池中已完成的任务总数，这同样是一个估计值；
  • getLargestPoolSize，线程池曾经创建过的最大线程数量。通过这个数据可以知道线程池是否充满过，也就是达到过maximumPoolSize；
  • getPoolSize，线程池当前的线程数量；
  • getActiveCount，当前线程池中正在执行任务的线程数量估计值。

四种常用线程池

很多情况下我们也不会直接创建ThreadPoolExecutor类的对象，而是根据需要通过Executors的几个静态方法来创建特定用途的线程池。目前常用的线程池有四种：

  1. 可缓存线程池，使用Executors.newCachedThreadPool方法创建
  2. 定长线程池，使用Executors.newFixedThreadPool方法创建
  3. 延时任务线程池，使用Executors.newScheduledThreadPool方法创建
  4. 单线程线程池，使用Executors.newSingleThreadExecutor方法创建

下面通过这些静态方法的源码来具体了解一下不同类型线程池的特性与适用场景。

可缓存线程池

JDK中的源码我们通过在IDE中进行跳转可以很方便地进行查看，下面就是Executors.newCachedThreadPool方法中的源代码。从代码中我们可以看到，可缓存线程池其实也是通过直接创建ThreadPoolExecutor类的构造器创建的，只是其中的参数都已经被设置好了，我们可以不用做具体的设置。所以我们要观察的重点就是在这个方法中具体产生了一个怎样配置的ThreadPoolExecutor对象，以及这样的线程池适用于怎样的场景。

从下面的代码中，我们可以看到，传入ThreadPoolExecutor构造器的值有： - corePoolSize核心线程数为0，代表线程池中的线程数可以为0 - maximumPoolSize最大线程数为Integer.MAX_VALUE，代表线程池中最多可以有无限多个线程 - 超时时间设置为60秒，表示线程池中的线程在空闲60秒后会被回收 - 最后传入的是一个SynchronousQueue类型的阻塞队列，代表每一个新添加的任务都要马上有一个工作线程进行处理

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

所以可缓存线程池在添加任务时会优先使用空闲的线程，如果没有就创建一个新线程，线程数没有上限，所以每一个任务都会马上被分配到一个工作线程进行执行，不需要在阻塞队列中等待；如果线程池长期闲置，那么其中的所有线程都会被销毁，节约系统资源。

  • 优点
    • 任务在添加后可以马上执行，不需要进入阻塞队列等待
    • 在闲置时不会保留线程，可以节约系统资源
  • 缺点
    • 对线程数没有限制，可能会过量消耗系统资源
  • 适用场景
    • 适用于大量短耗时任务和对响应时间要求较高的场景

定长线程池

传入ThreadPoolExecutor构造器的值有:

  • corePoolSize核心线程数和maximumPoolSize最大线程数都为固定值nThreads，即线程池中的线程数量会保持在nThreads，所以被称为“定长线程池”
  • 超时时间被设置为0毫秒，因为线程池中只有核心线程，所以不需要考虑超时释放
  • 最后一个参数使用了无界队列，所以在所有线程都在处理任务的情况下，可以无限添加任务到阻塞队列中等待执行
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

定长线程池中的线程数会逐步增长到nThreads个，并且在之后空闲线程不会被释放，线程数会一直保持在nThreads个。如果添加任务时所有线程都处于忙碌状态，那么就会把任务添加到阻塞队列中等待执行，阻塞队列中任务的总数没有上限。

  • 优点
    • 线程数固定，对系统资源的消耗可控
  • 缺点
    • 在任务量暴增的情况下线程池不会弹性增长，会导致任务完成时间延迟
    • 使用了无界队列，在线程数设置过小的情况下可能会导致过多的任务积压，引起任务完成时间过晚和资源被过度消耗的问题
  • 适用场景
    • 任务量峰值不会过高，且任务对响应时间要求不高的场景

延时任务线程池

与之前的两个方法不同，Executors.newScheduledThreadPool返回的是ScheduledExecutorService接口对象，可以提供延时执行、定时执行等功能。在线程池配置上有如下特点：

  • maximumPoolSize最大线程数为无限，在任务量较大时可以创建大量新线程执行任务
  • 超时时间为0，线程空闲后会被立即销毁
  • 使用了延时工作队列，延时工作队列中的元素都有对应的过期时间，只有过期的元素才会被弹出
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

延时任务线程池实现了ScheduledExecutorService接口，主要用于需要延时执行和定时执行的情况。

单线程线程池

单线程线程池中只有一个工作线程，可以保证添加的任务都以指定顺序执行（先进先出、后进先出、优先级）。但是如果线程池里只有一个线程，为什么我们还要用线程池而不直接用Thread呢？这种情况下主要有两种优点：一是我们可以通过共享的线程池很方便地提交任务进行异步执行，而不用自己管理线程的生命周期；二是我们可以使用任务队列并指定任务的执行顺序，很容易做到任务管理的功能。

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

总结

在这篇文章中我们从线程池的概念和基本使用方法说起，通过execute方法的源码深入剖析了任务提交的全过程和各个线程池构造器参数在线程池实际运行过程中所发挥的作用，还真正阅读了线程池类ThreadPoolExecutor的execute方法的源代码。最后，我们介绍了线程池的其他常用操作和四种常用的线程池。

到这里我们的线程池源代码之旅就结束了，希望大家在看完这篇文章之后能对线程池的使用和运行流程有了一个大概的印象。为什么说只是有了一个大概的印象呢？因为我觉得很多没有相关基础的读者读到这里可能还只是对线程池有了一个自己的认识，对其中的一些细节可能还没有完全捕捉到。所以我建议大家在看完这篇文章后不妨再返回到文章的开头多读几遍，相信第二遍的阅读能给大家带来不一样的体验，因为我自己也是在第三次读ThreadPoolExecutor类的源代码时才真正打通了其中的一些重要关节的。

引子

在这篇文章中，我们还只是探究了线程池的基本使用方法，以及提交任务方法execute的源代码。那么在任务提交以后是怎么被线程池所执行的呢？在下一篇文章中我们就可以找到答案，在下一篇文章中，我们会深入剖析线程池的任务执行流程。

avatar
文章
阅读
粉丝
相关推荐
Java 线程池
1.1k阅读
 · 
2点赞
剖析Tomcat线程池与JDK线程池的区别和联系！
3.7k阅读
 · 
93点赞
线程池不香了，结构化并发才是王道
36k阅读
 · 
342点赞
java线程-如何正确使用java线程池
4.8k阅读
 · 
41点赞
Java线程池详解二：线程池的使用
2.2k阅读
 · 
14点赞