理解Java并发工具包线程池的设计

一时失言乱红尘 2022-05-13 12:28 197阅读 0赞

### 为什么需要线程池？ ###

答：主要原因是因为创建一个线程开销太大，尤其是对大量的小任务需要执行这种场景。

在Java里面创建一个线程，需要包含的东西：

（1）它为一个线程堆栈分配内存，该堆栈为每个线程方法调用保存一个帧

（2）每个帧由局部变量数组，返回值，操作数栈，常量池组成

（3）某些JVM会为本地方法分配一个本地栈

（4）每个线程有一个程序计数器，用来告诉进程当前的指令执行到什么地方

（5）操作系统创建一个本机线程与java线程相对应

（6）文件描述符需要被创建，初始化然后添加到JVM内部的数据结构里面

（7）线程共享堆和方法区的内存

创建线程的流程依赖底层的操作系统，不同的操作系统可能不一样，此外更多的线程意味着 OS调度需要做更多的工作来决定哪一个线程可以访问资源，并且要通过OS调度切换维护线程的各种状态。

### 线程池的优点 ###

（1）降低资源消耗。通过重复利用已经创建的线程来降低各种资源消耗包括（线程的创建，销毁，状态的切换）

（2）提高响应速度。请求或者任务到达时直接处理。

（3）将任务的提交与任务执行分离，降低耦合。

（4）提高线程的可管理性。 使用线程池进行资源的统一分配，调优和监控。

### Java线程池的相关设计 ###

![70][]

线程池有关的接口和类

Java并发包在Java语言层面实现了自己的线程池，抽象封装了线程池的相关内容，从而可以做到更细粒度的资源控制：

与线程池相关的接口和类如下：

Executor接口：一个接口仅仅包含一个方法execute用来执行Runnable任务，主要定义了：

（1）通过这个接口就可以实现将任务提交和任务运行解耦，包括线程的详细使用，调度任务等等。

ExecutorService接口继承Executor接口：这个接口的主要定义了

（1）线程的线程池的关闭策略

shutdown() 告诉线程池，不能在接受新的任务，但是已经提交的任何或在等待执行的任务会继续运行

shutdownNow() 直接发送打断信号，让线程优雅的停止，如果忽略了中断信号，那么这个方法和shutdown方法作用一样

线程池关闭后，不会有任务还在执行，也不会有任务在等待执行，并且也不会有新的任务可以被提交，对于 不使用的 ExecutorService我们应该将其关闭，并回收其资源

（2）可以产生一个Future接口，用来跟踪一个或多个异步任务的运行进展

这个接口的submit方法，相当于是对Executor.execute(Runnable)接口方法的扩展，这个方法在提交任务之后，可以返回 一个Future接口，用于取消任务或者等待完成。 此外这个接口的invokeAny方法和invokeAll 方法可以用来执行一批任务， 然后等待至少一个或者全部任务完成。

最后这个接口还有一个awaitTermination方法，因为shutDown方法执行后，并不会阻塞到完成，所以我们可以使用这个方法来阻塞指定的时间，如果没有终止，就可以使用shutdownNow来发送打算信号，然后继续阻塞等待一定的时间，如果还没有终止，在指定的超时后，可以采用其他的办法，如强制退出虚拟机等，其间如果自身被打断，可以捕捉中断异常，再次关闭线程池，如果直到正常关闭后，保留中断的信号。

ScheduledExecutorService接口继承了ExecutorService接口，除了拥有父接口的定义外，该接口主要定义了：

可以调度任务在指定的延迟后执行一次或者周期性执行。

schedule方法可以创建不同的延迟任务，并返回一个task对象可以用来取消任务或者检查执行。

scheduleAtFixedRate和scheduleWithFixedDelay可以用来创建周期性的调度任务，直到被取消。

前者是每隔固定的延迟时间执行，后者也是间隔固定的时间，但是受前一个任务的完成时间影响，只有前者完成了后者才能执行。

AbstractExecutorService抽象类继承了ExecutorService接口：

提供了ExecutorService接口方法submit, invokeAny 和 invokeAll的默认实现，并对每一个执行任务的Runnable线程通过newTaskFor 方法产生了对应的RunnableFuture返回值用来跟踪任务的运行情况。

类图如下：

线程池的核心属性

ThreadPoolExecutor线程池的核心类继承了AbstractExecutorService类：

这个类定义了线程池核心参数和配置：

corePoolSize：线程池里面的最小的保持存活的worker数量，不允许超时，除非设置了allowCoreThreadTimeOut，最小是0；可以被动态改变

maximumPoolSize：线程池的最大数量，注意这个取决于阻塞队列的大小。可以被动态设置

keepAliveTime：线程的保持存活时间，如果超过这个时间值，没有任务提交就关闭自己，默认情况下 核心线程是不受这个参数影响的，除非设置了allowCoreThreadTimeOut=true。

threadFactory：除非使用者实现自己的线程工厂类，否则新线程的创建使用Executors.defaultThreadFactory()，通过默认工厂 创建出来的线程具有同样的组，优先级，和非守护进程的状态。提供一个不同的实现你可以修改，线程的名字，线程的组， 优先级，守护状态等。

handler：当阻塞队列满了之后的，制定的拒绝策略。

ThreadPoolExecutor类的主要构造方法如下：

new ThreadPoolExecutor
    (
    int corePoolSize, // 核心线程池的数量
    int maximumPoolSize，//最大线程池的数量
    long keepAliveTime, // 线程长期不用时，最大存活时间
    TimeUnit unit,   // 指定存活时间的单位
    BlockingQueue<Runnable> workQueue, //指定的阻塞队列
    ThreadFactory threadFactory, //指定的线程工厂类
    RejectedExecutionHandler handler //如果队列满了使用的拒绝策略
    )

参数比较多，所以在Executors类里面通过静态工厂方法，已经给我们涉及好了几种实现，我们可以 直接使用，下面我们详细看下最常见的几种线程池，如下：

（1）固定数量的线程池

拥有固定数量的线程来处理任务，如果全部都在处理任务， 新来的任务将会进入阻塞队列等待执行

newFixedThreadPool(int nThreads)//固定数量的线程池数
    ###这个方法的实际底层参数如下
    corePoolSize=maximumPoolSize，//核心数量=最大线程池数量=固定数量
    keepAliveTime=0//永远不销毁线程，等于0的情况，核心线程在不使用的时候不能被销毁
    unit=TimeUnit.MILLISECONDS//毫秒单位
    workQueue=new LinkedBlockingQueue<Runnable>() //无界阻塞队列
    threadFactory=new DefaultThreadFactory()//默认的工厂
    handler =new AbortPolicy() //队列满了，直接抛出异常

（2）单个线程的线程池

只有1个线程来处理任务，如果这个线程正在处理任务，新来的任务将会进入阻塞队列等待执行

newSingleThreadExecutor()//只创建一个线程来执行任务
    ###这个方法的实际底层参数如下
    corePoolSize=maximumPoolSize=1，//核心数量=最大线程池数量=1
    keepAliveTime=0//永远不销毁线程，等于0的情况，核心线程在不使用的时候不能被销毁
    unit=TimeUnit.MILLISECONDS//毫秒单位
    workQueue=new LinkedBlockingQueue<Runnable>() //无界阻塞队列
    threadFactory=new DefaultThreadFactory()//默认的工厂
    handler =new AbortPolicy() //队列满了，直接抛出异常

（3）拥有缓存效果的线程池

这个线程池在大量的执行时间短的异步任务时候，性能很高，内部的用的队列是Synchronous通过一对一 直接传递，可快速处理请求，并且处理完任务的线程会被缓存60秒，期间如果还有新任务到来，可以 复用先前的线程直接来处理，如果当前没有任务，那么这些线程超过60秒后会自动终止。

newSingleThreadExecutor()//只创建一个线程来执行任务
    ###这个方法的实际底层参数如下
    corePoolSize=0//不维护核心线程
    maximumPoolSize=Integer.MAX_VALUE，//允许最大的线程数
    keepAliveTime=60//超过60秒自动销毁
    unit=TimeUnit.SECONDS//单位秒
    workQueue=new SynchronousQueue<Runnable>()) //无界阻塞队列
    threadFactory=new DefaultThreadFactory()//默认的工厂
    handler =new AbortPolicy() //队列满了，直接抛出异常

（4）延迟指定时间后执行一次任务或者周期执行任务的只拥有单个线程的线程池

这个线程池其实是ThreadPoolExecutor的子类ScheduledThreadPoolExecutor提供的功能：

newScheduledThreadPool//只创建一个线程来执行任务
    ###这个方法的实际底层参数如下
    corePoolSize=1//核心线程只有一个
    maximumPoolSize=Integer.MAX_VALUE，//允许最大的线程数
    keepAliveTime=0//永远不销毁线程，等于0的情况，核心线程在不使用的时候不能被销毁
    unit=TimeUnit.NANOSECONDS//单位纳秒
    workQueue=new DelayedWorkQueue() //类似DelayQueue队列的，内部采用堆结构实现
    threadFactory=new DefaultThreadFactory()//默认的工厂
    handler =new AbortPolicy() //队列满了，直接抛出异常

（5）延迟指定时间后执行一次任务或者周期执行任务的只拥有执行线程的线程池

这个线程池其实是ThreadPoolExecutor的子类ScheduledThreadPoolExecutor提供的功能：

newScheduledThreadPool//只创建一个线程来执行任务
    ###这个方法的实际底层参数如下
    corePoolSize=n//指定的数量
    maximumPoolSize=Integer.MAX_VALUE，//允许最大的线程数
    keepAliveTime=0//永远不销毁线程，等于0的情况，核心线程在不使用的时候不能被销毁
    unit=TimeUnit.NANOSECONDS//单位秒
    workQueue=new DelayedWorkQueue() //类似DelayQueue队列的，内部采用堆结构实现
    threadFactory=new DefaultThreadFactory()//默认的工厂
    handler =new AbortPolicy() //队列满了，直接抛出异常

（6）ForkJoinPool基于工作窃取算法（ work-stealing）的线程池

这个线程池其实是ThreadPoolExecutor的子类ForkJoinPool类提供的功能：

newWorkStealingPool(nThreads) //默认获取服务器的cpu的个数，也可以自己指定并行度
    ###这个方法的实际底层参数如下
    
    parallelism=n // 并行任务个数
    factory=ForkJoinWorkerThreadFactory  //forkjoin的默认工厂
    handler=null //不存在拒绝策略
    asyncMode （true=FIFO先进先出模式处理队列任务，false=LIFO后进先出的模式处理任务）
    workerNamePrefix  //线程组的前缀

工作窃取（work-stealing）算法是指某个线程从其他队列里窃取任务来执行。那么，为什么 需要使用工作窃取算法呢？假如我们需要做一个比较大的任务，可以把这个任务分割为若干 互不依赖的子任务，为了减少线程间的竞争，把这些子任务分别放到不同的队列里，并为每个 队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应。比如A线程负责处理A 队列里的任务。但是，有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有 任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列 里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被 窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿 任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。

![image][]

线程池的阻塞队列

queue：阻塞队列BlockingQueue接口可以用来转换和保存提交的任务，

当一个任务来的时候，如果当前的线程数小于corePoolSize，Executors会新建线程来处理，即使有其他的 空闲线程，如果corePoolSize满了，Executors把新的任务添加到阻塞队列里面，而不是新创建线程，如果队列 也满了，但是当前的任务总数小于maximumPoolSize，那么新的线程会被创建，如果总个数大于maximumPoolSize，那么 新的任务会被拒绝。

如下图示：

![70 1][]

阻塞队列主要包括三种：

（1）直接交付 （SynchronousQueue）

（2）无边界队列（LinkedBlockingQueue）

（3）有边界队列 （ArrayBlockingQueue）

workerCount的值2的29次方-1约500万，代表这个线程池总共出现执行过的线程数量

线程池的状态

runState：代表线程池的状态

（1）RUNNING 当前接受新任务提交，并且也处理队列里面的任务

（2）SHUTDOWN 不接受新任务 ， 但是处理队列里面的任务

（3）STOP 不接受新任务，不处理队列任务，并且打断正在运行的任务

（4）TIDYING 所有的任务已经终止，workerCount是0，此时线程的状态切为TIDYING， 结束时可以运行钩子方法terminated()

（5）TERMINATED terminated方法调用完成。

状态转变：

RUNNING => SHUTDOWN 调用shutdown方法

RUNNING 或者 SHUTDOWN => 调用shutdownNow方法

SHUTDOWN => TIDYING 当队列和线程池都为空的情况下

STOP => TIDYING 当线程池是空的时候。

TIDYING =>TERMINATED 当terminated()钩子方法调用完成的时候。

另外注意如果调用了awaitTermination方法，只有当线程池的状态 转变为TERMINATED的时候，这个方法才会返回。

线程池的拒绝策略

关于线程池如果满了或者线程池已经关闭时的拒绝策略：

AbortPolicy：直接抛出一个RejectedExecutionException异常

CallerRunsPolicy：直接使用调用者线程用来执行任务，提供了一个简单的反压控制机制 ，从而降低新任务提交的速度。

DiscardPolicy：直接丢弃任务。

DiscardOldestPolicy： 如果executor没有关闭，丢弃头部的任务，然后执行重试 ，如果重试失败，则重复尝试。

钩子方法：

beforeExecute(Thread, Runnable)

afterExecute(Runnable, Throwable)

可以在线程的执行前后做一些处理，比如添加log日志， 收集统计数据等。 此外还可以覆盖重写terminated()，在线程池彻底关闭时做一些处理工作。

关于队列的维护：

ThreadPoolExecutor类提供了一个getQueue()方法，允许访问当前的工作队列去监控和调试，此外 这个类还有两个方法：remove(Runnable) 和 purge() 用来删除或者取消任务来辅助回收资源的， 不建议使用这些方法。

资源释放

如果一个线程池长时间不再活动，或者用户忘记调用shutdown方法，我们也希望回收资源，这个时候我们可以设置 allowCoreThreadTimeOut(boolean)这个参数使得核心线程也能在长时间不用时回收掉。

异步任务的获取Future

Future用来代表对于异步任务的结果，可以通过Future来检查任务是否执行完成， 以及阻塞等待其完成，获取结算结果，取消任务（已经完成的任务不能取消）等

### 总结 ###

本篇文章主要了Java线程池的出现的意义及Java线程池的相关设计与相关内容的概述，通过线程池我们可以将任务的提交与执行分离，从而降低与程序的耦合，此外利用线程池我们还可以降低资源的消耗，提高线程的可管理性，进行资源的统一分配，调优和监控。

[70]: /images/20220513/5a0419a5e4c14a4bbe0e7f51fdff8d25.png
[image]: /images/20220513/6f4ec69fc2034f3c82b1821f2847fe10.png
[70 1]: /images/20220513/527f5a143f9144b6bea22020d155add5.png