Java并发-线程池篇-附场景分析

Posted 汤圆学Java

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Java并发-线程池篇-附场景分析相关的知识，希望对你有一定的参考价值。

作者：汤圆

个人博客：javalover.cc

前言

前面我们在创建线程时，都是直接new Thread()；

这样短期来看是没有问题的，但是一旦业务量增长，线程数过多，就有可能导致内存异常OOM，CPU爆满等问题

幸运的是，Java里面有线程池的概念，而线程池的核心框架，就是我们今天的主题，Executor

接下来，就让我们一起畅游在Java线程池的海洋中吧

本节会用银行办业务的场景来对比介绍线程池的核心概念，这样理解起来会很轻松

简介

Executor是线程池的核心框架；

和它相对应的有一个辅助工厂类Executors，这个类提供了许多工厂方法，用来创建各种各样的线程池，下面我们先看下几种常见的线程池

// 容量固定的线程池
Executor fixedThreadPool = Executors.newFixedThreadPool(5);
// 容量动态增减的线程池
Executor cachedThreadPool = Executors.newCachedThreadPool();
// 单个线程的线程池
Executor singleThreadExecutor = Executors.newSingleThreadExecutor();
// 基于调度机制的线程池（不同于上面的线程池，这个池创建的任务不会立马执行，而是定期或者延时执行）
Executor scheduledThreadPool = Executors.newScheduledThreadPool(5);

上面这些线程池的区别主要就是线程数量的不同以及任务执行的时机

下面让我们开始吧

文章如果有问题，欢迎大家批评指正，在此谢过啦

目录

  1. 线程池的底层类ThreadPoolExecutor
  2. 为啥阿里不建议使用 Executors来创建线程池？
  3. 线程池的生命周期 ExecutorService

正文

1. 线程池的底层类 ThreadPoolExecutor

在文章开头创建的几个线程池，内部都是有调用ThreadPoolExecutor这个类的，如下所示

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

这个类是Exexutor的一个实现类，关系图如下所示：

  • 其中Executors就是上面介绍的辅助工厂类，用来创建各种线程池

  • 接口ExecutorService是Executor的一个子接口，它对Executor进行了扩展，原有的Executor只能执行任务，而ExecutorService还可以管理线程池的生命周期（下面会介绍）

所以我们先来介绍下这个底层类，它的完整构造参数如下所示：

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

在介绍这些参数之前，我们可以先举个生活中的例子-去银行办业务；然后对比着来理解，会比较清晰

（图中绿色的窗口表示一直开着）

  • corePoolSize： 核心线程数，就是一直存在的线程（不管用不用）；=》窗口的1号窗和2号窗
  • maximumPoolSize：最大线程数，就是最多可以创建多少个线程；=》窗口的1，2，3，4号窗
  • keepAliveTime：多余的线程（最大线程数 减去 核心线程数）空闲时存活的时间；=》窗口的3号窗和4号窗空闲的时间,如果超过keepAliveTime，还没有人来办业务，那么就会暂时关闭3号窗和4号窗
  • workQueue: 工作队列，当核心线程数都在执行任务时，再进来的任务就会添加到工作队列中；=》椅子，客户等待区
  • threadFactory：线程工厂，用来创建初始的核心线程，下面会有介绍；
  • handler：拒绝策略，当所有线程都在执行任务，且工作队列也满时，再进来的任务就会被执行拒绝策略（比如丢弃）；=》左下角的那个小人

基本的工作流程如下所示：

上面的参数我们着重介绍下工作队列和拒绝策略，线程工厂下面再介绍

工作队列：

  • ArrayBlockingQueue：
    • 数组阻塞队列，这个队列是一个有界队列，遵循FIFO，尾部插入，头部获取
    • 初始化时需指定队列的容量 capacity
    • 类比到上面的场景，就是椅子的数量为初始容量capacity
  • LinkedBlockingQueue：
    • 链表阻塞队列，这是一个无界队列，遵循FIFO，尾部插入，头部获取
    • 初始化时可不指定容量，此时默认的容量为Integer.MAX_VALUE，基本上相当于无界了，此时队列可一直插入（如果处理任务的速度小于插入的速度，时间长了就有可能导致OOM)
    • 类比到上面的场景，就是椅子的数量为Integer.MAX_VALUE
  • SynchronousQueue：
    • 同步队列，阻塞队列的特殊版，即没有容量的阻塞队列，随进随出，不做停留
    • 类比到上面的场景，就是椅子的数量为0，来一个人就去柜台办理，如果柜台满了，就拒绝
  • PriorityBlockingQueue
    • 优先级阻塞队列，这是一个无界队列，不遵循FIFO，而是根据任务自身的优先级顺序来执行
    • 初始化可不指定容量，默认11（既然有容量，怎么还是无界的呢？因为它添加元素时会进行扩容）
    • 类比到上面的场景，就是新来的可以插队办理业务，好比各种会员

拒绝策略：

  • AbortPolicy（默认）：
    • 中断策略，抛出异常 RejectedExecutionException；
    • 如果线程数达到最大，且工作队列也满，此时再进来任务，则抛出 RejectedExecutionException（系统会停止运行，但是不会退出）
  • DiscardPolicy：
    • 丢弃策略，丢掉新来的任务
    • 如果线程数达到最大，且工作队列也满，此时再进来任务，则直接丢掉（看任务的重要程度，不重要的任务可以用这个策略）
  • DiscardOldestPolicy：
    • 丢弃最旧策略，丢掉最先进入队列的任务（有点残忍了），然后再次执行插入操作
    • 如果线程数达到最大，且工作队列也满，此时再进来任务，则直接丢掉队列头部的任务，并再次插入任务
  • CallerRunsPolicy：
    • 回去执行策略，让新来的任务返回到调用它的线程中去执行（比如main线程调用了executors.execute(task)，那么就会将task返回到main线程中去执行）
    • 如果线程数达到最大，且工作队列也满，此时再进来任务，则直接返回该任务，到调用它的线程中去执行

2. 为啥阿里不建议使用 Executors来创建线程池？

原话如下：

我们可以写几个代码来测试一下

先测试FixedThreadPool，代码如下：

public class FixedThreadPoolDemo {
    public static void main(String[] args) {
        // 创建一个固定容量为10的线程池，核心线程数和最大线程数都为10
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 1_000_000; i++) {
            try{
                executorService.execute(()->{
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}

这里我们需对VM参数做一点修改，让问题比较容易复现

如下所示，我们添加-Xmx8m -Xms8m到VM option中(-Xmx8m：JVM堆的最大内存为8M， -Xms8m，JVM堆的初始化内存为8M)：

此时点击运行，就会发现报错如下：

Exception in thread "main" java.lang.OutOfMemoryError: Java heap space
	at java.util.concurrent.LinkedBlockingQueue.offer(LinkedBlockingQueue.java:416)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1371)
	at com.jalon.concurrent.chapter6.FixedThreadPoolDemo.main(FixedThreadPoolDemo.java:21)

我们来分析下原因

  • 首先，newFixedThreadPool内部用的工作队列为LinkedBlockingQueue，这是一个无界队列（容量最大为Integer.MAX_VALUE，基本上可一直添加任务）
  • 如果任务插入的速度，超过了任务执行的速度，那么队列肯定会越来越长，最终导致OOM

CachedThreadPool也是类似的原因，只不过它是因为最大线程数为Integer.MAX_VALUE；

所以当任务插入的速度，超过了任务执行的速度，那么线程的数量会越来越多，最终导致OOM

那我们要怎么创建线程池呢？

可以用ThreadPoolExecutor来自定义创建，通过为最大线程数和工作队列都设置一个边界，来限制相关的数量，如下所示：

public class ThreadPoolExecutorDemo {
    public static void main(String[] args) {
        ExecutorService service = new ThreadPoolExecutor(
                1, // 核心线程数
                1, // 最大线程数
                60L, // 空闲时间
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1), // 数组工作队列，长度1
                new ThreadPoolExecutor.DiscardPolicy()); // 拒绝策略：丢弃
        for (int i = 0; i < 1_000_000; i++) {
            // 通过这里的打印信息，我们可以知道循环了3次
            // 原因就是第一次的任务在核心线程中执行，第二次的任务放到了工作队列，第三次的任务被拒绝执行
            System.out.println(i);
            service.execute(()->{
                // 这里会报异常，是因为执行了拒绝策略（达到了最大线程数，队列也满了，此时新进来的任务就会执行拒绝策略）
                // 这里需要注意的是，抛出异常后，代码并不会退出，而是卡在异常这里，包括主线程也会被卡住(这个是默认的拒绝策略）
                // 我们可以用其他的拒绝策略，比如DiscardPolicy,此时代码就会继续往下执行
                System.out.println(Thread.currentThread().getName());
            });
        }
        try {
            Thread.sleep(1000);
            System.out.println("主线程 sleep ");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

3. 线程池的生命周期 ExecutorService

Executor接口默认只有一个方法void execute(Runnable command);，用来执行任务

任务一旦开启，我们就无法再去插手了，比如停止、监控等

此时就需要ExecutorService登场了，它是Executor的一个子接口，对其进行了扩展，方法如下：

public interface ExecutorService extends Executor {

    void shutdown(); // 优雅地关闭，这个关闭会持续一段时间，以等待已经提交的任务去执行完成（但是在shutdown之后提交的任务会被拒绝）

    List<Runnable> shutdownNow(); // 粗暴地关闭，这个关闭会立即关闭所有正在执行的任务，并返回工作队列中等待的任务

    boolean isShutdown();

    boolean isTerminated();

    // 用来等待线程的执行
    // 如果在timeout之内，线程都执行完了，则返回true；
    // 如果等了timeout，还没执行完，则返回false；
    // 如果timeout之内，线程被中断，则抛出中断异常
    boolean awaitTermination(long timeout, TimeUnit unit) 
        throws InterruptedException;

    <T> Future<T> submit(Callable<T> task);
   
    <T> Future<T> submit(Runnable task, T result);
}

从上面可以看到，线程池的生命周期分三步：

  1. 运行：创建后就开始运行
  2. 关闭：调用shutdown进入关闭状态
  3. 已终止：所有线程执行完毕

总结

  1. 线程池的底层类 ThreadPoolExecutor：核心概念就是核心线程数、最大线程数、工作队列、拒绝策略
  2. 为啥阿里不建议使用 Executors来创建线程池？：因为会导致OOM，解决办法就是自定义ThreadPoolExecutor，为最大线程数和工作队列设置边界
  3. 线程池的生命周期ExecutorService：运行状态（创建后进入）、关闭状态（shutdown后进入）、已终止状态（所有线程都执行完成后进入）

参考内容：

  • 《Java并发编程实战》
  • 《实战Java高并发》
  • newFixedThreadPool的弊端：https://my.oschina.net/langwanghuangshifu/blog/3208320
  • 银行办业务的场景参考：https://b23.tv/ygGjTH

后记

愿你的意中人亦是中意你之人

以上是关于Java并发-线程池篇-附场景分析的主要内容，如果未能解决你的问题，请参考以下文章

Java并发-线程池篇-附场景分析

Java Semaphore实现高并发场景下的流量控制（附源码） | 实用代码架构

Java 并发编程线程操作原子性问题 ( 问题业务场景分析 | 使用 synchronized 解决线程原子性问题 )

Java 并发编程线程共享变量可见性 ( volatile 关键字使用场景分析 | MESI 缓存一致性协议 | 总线嗅探机制 )

干货：Java并发编程必懂知识点解析（内附面试题）

2020最新Java工程师面试题-Java 并发编程（附答案，持更中）