Java并发—Executor框架

叁歲伎倆 2024-04-18 14:01 76阅读 0赞

## Executor框架 ##

在Java中, 使用线程来异步执行任务. Java线程的创建与销毁需要一定的开销, 如果我们为每一个任务创建一个新线程来执行, 这些线程的创建与销毁将消耗大量的计算资源. 同时, 为每一个任务创建一个新线程来执行, 这种策略可能会使处于高负荷的应用最终崩溃.

### 1.Executor框架简介 ###

#### 1>Executor框架的两级调度模型 ####

在HotSpot VM的线程模型中, Java线程被**一对一**映射为本地操作系统线程(轻量级进程LWP). Java线程启动时会创建一个本地操作系统线程. 当该Java线程终止时, 这个操作系统线程也会被回收. 操作系统会调度所有线程并将他们分配给可用的CPU.

在上层, Java多线程程序通常把应用分解为若干个任务, 然后使用用户级的调度器(Executor框架)将这些任务映射为固定数量的线程; 在底层, 操作系统内核将这些线程映射到硬件处理器上. 应用通过Executor框架控制上层的调度; 而下层的调度由操作系统内核控制, 下层的调度不应受应用程序控制

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70]

#### 2>Executor框架的结构与成员 ####

Executor框架可分为两部分 : Executor的结构和Executor框架包含的成员组件

##### 1.Executor框架的结构 #####

Executor框架主要由3大部分组成如下 :

*  任务 : 包括被执行任务需要实现的接口 : Runnable接口或Callable接口
 *  任务的执行 : 包括任务执行机制的核心接口Executor, 以及继承自Executor的ExecutorService接口与.Executor有两个关键类实现了ExecutorService接口(ThreadPoolExecutor和ScheduledThreadExecutor)
 *  异步计算的结果 : 包括接口Future和实现Future接口的FutureTask类

下面是一些类和接口的简介 :

*  Executor是一个接口, 它是Executor框架的基础, 它将任务的提交与任务的执行分离开来
 *  ThreadPoolExecutor是线程池的核心实现类, 用来执行被提交的任务
 *  ScheduledThreadPoolExecutor是一个实现类, 可以在给定的延迟后运行命令, 或者定期执行命令. ScheduledThreadPoolExecutor比Timer更灵活, 功能更强大
 *  Future接口和实现Future接口的FutureTask类, 代表异步计算的结果
 *  Runnable接口Callable接口的实现类, 都可以被ThreadPoolExecutor和ScheduledThreadPoolExecutor执行

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 1]

主线程首先要创建Runnable或者Callable接口的任务对象. 工具类Executors可以把一个Runnable对象封装为一个Callable对象(Executors.callable(Runnable task))或Executors.callable(Runnable task, Object result))

然后可以把Runnable对象直接交给ExecutorService执行(ExecutorService.execute(Runnable command)); 或者也可以把Runnable对象或Callable对象提交给ExecutorService执行(ExecutorService.submit(Runnable task) 或ExecutorService.submit(Callable< T> task))

如果执行ExecutorService.submit(…), ExecutorService将返回一个实现Future接口的对象(现在的版本是FutureTask对象).

##### 2.Executor框架的成员 #####

*  ThreadPoolExecutor通常使用工厂类Executors来创建, Executors可以创建三种类型的ThreadPoolExecutor : SingleThreadExecutor, FixedThreadPool, CachedThreadPool
    
     *  FixedThreadPool : 使用固定线程数的线程池, 适用于为了满足资源管理的需求, 而需要限制当前线程数量的应用场景, 适用于负载比较重的服务器
     *  SingleThreadExecutor : 使用单个线程的SingleThreadExecutor. 适用于需要保证顺序地执行各个任务, 并且在任意时间不会有多个线程是活动的应用场景
     *  CachedThreadPool : 大小无界的线程池, 适用于很多短期异步任务的小程序, 或者是负载较清的服务器
 *  ScheduledThreadPoolExecutor通常使用Executors工厂类来创建, Executors可以创建两种类型的ScheduledThreadPoolExecutor
    
     *  ScheduledThreadPoolExecutor : 包含若干线程的ScheduledThreadPoolExecutor, 适用于需要多个后台线程执行的周期任务
     *  SingleScheduledThreadPoolExecutor : 只包含一个线程的ScheduledThreadPoolExecutor, 适用于单个后台线程执行周期任务
 *  Future接口
    
    Future接口和实现Future接口的FutureTask类用来表示**异步计算的结果**. 当把Runnable接口或Callable接口的实现类提交(submit)给ThreadPoolExecutor时, 会返回一个FutureTask对象
 *  Runnable接口和Callable接口
    
    Runnable接口和Callable接口的实现类, 都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor执行, 它们之间的区别是Runnable不会返回结果, 而Callable可以返回结果

### 2.ThreadPoolExecutor详解 ###

Executor框架最核心的类是ThreadPoolExecutor, 它是线程池的实现类, 主要由下面4个组件构成

*  corePool : 核心线程池的大小
 *  maximum : 最大线程池的大小
 *  BlockingQueue : 用来暂时保存任务的工作队列
 *  RejectedExecutionHandler : 当ThreadPoolExecutor已经关闭或ThreadPoolExecutor已经饱和(达到了最大线程池大小且工作队列已满), execute()方法将要调用的Handler

通过Executor框架的工具类Executors, 可以创建3种类型的ThreadPoolExecutor

*  FixedThreadPool
 *  SingleThreadExecutor
 *  CachedThreadPool

#### 1>FixedThreadPool ####

FixedThreadPool被称为**可重用固定线程数的线程池**. FixedThreadPool的corePoolSize和maximumPoolSize都被设置为创建FixedThreadPool时指定的参数nThreads; 当线程池中的线程数大于corePoolSize时, keepAliveTIme为多于空闲线程等待新任务的最长时间, 超过这个时间后多余的线程将被终止. 把keepAliveTime设置为0L就意味着多余的空闲线程会被立即终止

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 2]

*  1.如果当前运行的线程数少于corePoolSize, 则创建新线程来执行任务
 *  2.在线程池完成预热之后(当前运行的线程数等于corePoolSize), 将任务加入LinkedBlockingQueue
 *  3.线程执行完1中的任务后, 会在循环反复从LinkedBlockingQueue获取任务来执行

FixedThreadPool使用无界队列LinkedBlocking作为线程池的工作队列(队列的容量为Integer.MAX\_VALUE). 使用无界队列作为工作队列会对线程带来如下区别.

*  线程池中的线程数量不会超过corePoolSize
 *  maximumPoolSize是一个无效参数
 *  keepAliveTime是一个无效参数
 *  运行中的FixedThreadPool不会拒绝任务

#### 2>SingleThreadExecutor ####

SingleThreadExecutor是使用单个worker线程的Executor. SingleThreadExecutor的corePoolSize和maximum参数被设置为1. 其他参数与FixedThreadPool相同. SingleThreadExecutor使用无界队列LinkedBlockingQueue作为线程池的工作队列

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 3]

*  如果当前线程数小于corePoolSize(即线程池中无运行的线程, 则创建一个新线程来执行任务)
 *  在线程池完成预热之后(当前线程池中会有一个运行的线程), 将任务加入LinkedBlockingQueue
 *  线程执行完1中的任务后, 会在一个无限循环中反复从LinkedBlockingQueue中获取任务来执行

#### 3>CachedThreadPool ####

CachedThreadPool是一个会根据需要创建新线程的线程池. CachedThreadPool的corePoolSize被设置为0, 即corePool为空; maximumPoolSize被设置为Integer.MAX\_VALUE, 即maximumPool是无界的. 这里把keepAliveTime设置为60L, 也就是CachedThreadPool中的空闲线程等待新任务的最长时间为60秒. 空闲线程超过60秒后会被终止

FixedThreadPool和SingleThreadExecutor使用无界队列LinkedBlockingQueue作为现场的工作队列. CachedThreadPool使用没有容量的SynchronousQueue作为线程池的工作队列, 但CachedThreadPool中的maximumPool是无界的, 也就是说如果主线程提交任务的速度高于maximumPool中线程处理任务的速度时, CachedThreadPool会不断创建新线程. 极端情况下, CachedThreadPool会因为创建过多线程而耗尽CPU和内存资源

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 4]

*  1.首先执行SynchronousQueue.offer(Runnable task). 如果当前maximumPool中有空闲线程在执行SynchronousQueue.poll(keepAliveTime, TImeUnit.NANOSECONDS), 那么主线程执行的offer操作与空闲线程执行的poll操作配对成功
 *  2.当初始maximumPool为空, 或者maximum中当前没有空闲线程时, 将没有线程执行SynchronousQueue.poll(keeyAliveTime, TImeUnit.NANOSECONDS). 这种情况下, 步骤1将失败, 此时CachedThreadPool会创建一个新线程执行任务, execute()方法执行完成
 *  3.在步骤2中新创建的线程将任务执行完之后, 会执行SynchronousQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS). 这个poll操纵会让空闲线程最多在SynchronousQueue中等待60秒中. 如果60秒内主线程提交了一个新任务(主线程执行步骤1),那么这个空闲线程将执行主线程提交的新任务; 否则, 这个空闲线程将被终止.因此长时间保持空闲的CachedThreadPool不会使用任何资源

SynchronousQueue是一个没有容量的阻塞队列. 每个插入操作必须等待另一个线程对应的移除操作. CachedThreadPool使用SynchronousQueue, 把主线程提交的任务传递给空闲线程执行.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 5]

### 3.ScheduledThreadPoolExecutor ###

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor. 它主要用来在给定的延迟之后运行任务, 或者定期执行任务.ScheduledThreadPoolExecutor的功能与TImer类似, 但ScheduledThreadPoolExecutor功能更强大, 更灵活. TImer对应的是单个后台线程, 而ScheduledThreadPoolExecutor可以在构造函数中指定多个对应的后台线程数

#### 1>ScheduledThreadPoolExecutor运行机制 ####

ScheduledThreadPoolExecutor的执行示意图如图

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 6]  
DelayQueue是一个无界队列, 所以ThreadPoolExecutor的maximumPoolSize在ScheduledThreadPoolExecutor中没有什么意义.

ScheduledThreadPoolExecutor的执行主要分为两大部分

*  当调用ScheduledThreadPoolExecutor的scheduleAtFixedRate()方法或者添加scheduleWithFixedDelay方法时, 会向ScheduledThreadPoolExecutor的DelayQueue添加一个实现了RunnableScheduledFutur接口的ScheduledFutureTask
 *  线程池中的线程从DelayQueue中获取ScheduledFutureTask, 然后执行任务

#### 2>ScheduledThreadPoolExecutor的实现 ####

ScheduledThreadPoolExecutor会把待调度的任务(ScheduledFutureTask)放到一个DelayQueue中, ScheduledFutureTask主要包含3个成员变量, 如下 :

*  long型成员变量time, 表示这个任务将要被执行的具体时间
 *  long型成员变量sequenceNumber, 表示这个任务被添加到ScheduledThreadPoolExecutor中的序号
 *  long型成员变量period,表示任务执行的间隔周期

DelayQueue封装了一个PriorityQueue, 这个PriorityQueue会对队列中的ScheduledFutureTask进行排序. 排序时, time小的排在前面(时间早的任务先被执行). 如果两个ScheduledFutureTask的time相同, 就比较sequenceNumber, sequenceNumber小的排在前面(先提交的任务先执行)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 7]

*  1.线程1从DelayQueue中获取到已到期的ScheduledFutureTask(DelayQueue.take()). 到期任务是指ScheduledFutureTask的time大于等于当前时间
 *  2.线程1执行这个ScheduledFutureTask
 *  3.线程1修改ScheduledFutureTask的time变量为下次将要执行的时间
 *  4.线程1把这个修改time之后的ScheduledFutureTask放回DelayQueue中(DelayQueue.add())

### 4.FutureTask ###

#### 1>FutureTask简介 ####

FutureTask除了实现Future接口之外, 还实现了Runnable接口. 因此, FutureTask可以交给Executor执行, 也可以由调用线程直接执行.根据FutureTask.run()方法被执行的时机, FutureTask可以处于下面3种状态

*  1.未启动, FutureTask.run()方法还没有执行之前, FutureTask处于未启动状态. 当创建一个FutureTask, 且没有执行FutureTask.run()方法之前, 这个FutureTask处于未启动状态
 *  2.已启动, FutureTask.run()方法被执行的过程中, FutureTask处于已启动状态.
 *  3.已完成, FutureTask.run()方法执行完后正常结束, 或被取消(FutureTask.cancel(…)), 或执行FutureTask.run()方法时抛出异常而异常结束, FutureTask处于已完成状态

当FutureTask处于未启动或已启动状态时, 执行FutureTask.get()方法将导致调用线程阻塞.当FutureTask处于已完成状态时, 执行FutureTask.get()方法将导致调用线程立即返回结果或抛出异常

#### 2>FutureTask的使用 ####

可以把FutureTask交给Executor执行; 也可以通过ExecutorService.submit(…)方法返回一个FutureTask, 然后执行FutureTask.get()或FutureTask.cancel()方法. 还可以单独使用FutureTask

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 8]

#### 3>FutureTask的实现 ####

FutureTask的实现是基于AbstractQueuedSynchronizer(AQS). J.U.C中很多可阻塞类都是基于AQS实现的. AQS是一个同步框架, 它提供通用原子性管理同步状态, 阻塞和唤醒线程, 以及维护被阻塞线程的队列.

每个基于AQS实现的同步器都会包含两种类型的操作 :

*  至少一个acquire操作. 这个操作阻塞调用线程, 除非/直到AQS的状态允许这个线程继续执行.
 *  至少一个release操作. 这个操作改变AQS的状态, 改变后的状态可允许一个或多个阻塞线程被接触阻塞

FutureTask声明了一个内部私有的继承于AQS的内部子类Sync, 这个内部子类只用实现状态检查和状态更新的方法即可, 这些方法将控制FutureTask的获取和释放操作. Sync实现了AQS的tryAcquireShared(int)方法和tryReleaseShared(int)方法, Sync通过这两个方法来检查和更新同步状态

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 9]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/fc9865f643d444cd88392c72da21ced9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/34cd985403074c62ae5d0158eb43acad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/9288ac038ac3473ca5c064fc772a629a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/ca4abe1c50ac46bf9eff5ce143197bbc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/6249ace0cae64fc796e89a72a8ceef49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/fcb882e734624e33a43b803e3b88dc8a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/ea645593192b4ecb9cd2ed0f33599ac9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 7]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/bd05fadfbf5345f6a8baf1e62f79e1ef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 8]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/cac377da2cde47feb55b26a37122ba60.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbnRlcnNoaWk_size_16_color_FFFFFF_t_70 9]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/2e45d3ebb6df4814ada2f7267abdc809.png