Executor框架

客官°小女子只卖身不卖艺 2022-10-20 01:14 159阅读 0赞

# 0.Executor框架的结构 #

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

*  **任务**：包括被执行任务需要实现的接口：Runnable接口或Callable接口。
 *  **任务的执行**：包括任务执行机制的核心接口Executor，以及继承自Executor的 ExecutorService接口。Executor框架有两个关键类实现了ExecutorService接口 （ThreadPoolExecutor和ScheduledThreadPoolExecutor）。
 *  **异步计算的结果**。包括接口Future和实现Future接口的FutureTask类。

下面是这些类和接口的简介

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

**使用Executor框架的大致流程**：

1.  创建一个实现了Runnable或者Callable接口的对象。
2.  使用Executor的execut方法执行。或者ExecutorService的submit方法
3.  FutureTask.get()获取异步任务返回的结果

# 1.ThreadPoolExecutor #

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

*  corePool：核心线程池的大小
 *  maximumPool：最大线程池的大小
 *  BlockingQueue：用来暂时保存任务的工作队列。
 *  RejectedExecutionHandler：拒绝策略

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

*  FixedThreadPool。
 *  SingleThreadExecutor。
 *  CachedThreadPool。

## 1.1 FixedThreadPool ##

FixedThreadPool被称为可重用固定线程数的线程池

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

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

FixedThreadPool使用无界队列LinkedBlockingQueue作为线程池的工作队列。  
无界队列作为工作队列会对线程池带来如下影响：

1.  当线程池中的线程数达到corePoolSize后，新任务将在无界队列中等待，因此线程池中 的线程数不会超过corePoolSize。
2.  由于1，使用无界队列时maximumPoolSize将是一个无效参数
3.  由于1和2，使用无界队列时keepAliveTime将是一个无效参数
4.  由于使用无界队列，运行中的FixedThreadPool（未执行方法shutdown()或 shutdownNow()）不会拒绝任务（不会调用RejectedExecutionHandler.rejectedExecution方法）。

## 1.2 SingleThreadExecutor ##

SingleThreadExecutor是使用单个worker线程的Executor。

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

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

SingleThreadExecutor使用无界队列LinkedBlockingQueue作为线程池的工 作队列（队列的容量为Integer.MAX\_VALUE）。

## 1.3 CachedThreadPool ##

CachedThreadPool是一个会根据需要创建新线程的线程池

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

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

CachedThreadPool的corePoolSize被设置为0，即corePool为空；maximumPoolSize被设置为 Integer.MAX\_VALUE，即maximumPool是无界的。这里把keepAliveTime设置为60L，意味着 CachedThreadPool中的空闲线程等待新任务的最长时间为60秒，空闲线程超过60秒后将会被终止。使用没有容量的SynchronousQueue作为线程池的工作队列

这意味着，如果主线程提交任务的速度高于 maximumPool中线程处理任务的速度时，CachedThreadPool会不断创建新线程。极端情况下， CachedThreadPool会因为创建过多线程而耗尽CPU和内存资源。

# 2.ScheduledThreadPoolExecutor #

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor。它主要用来在给定的延迟之后运 行任务，或者定期执行任务。  
DelayQueue是一个无界队列，所以ThreadPoolExecutor的maximumPoolSize在Scheduled- ThreadPoolExecutor中没有什么意义。DelayQueue封装了一个PriorityQueue，这个PriorityQueue会对队列中的ScheduledFutureTask进行排序

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

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

ScheduledFutureTask有3个成员变量

// 任务被添加进executor中的序号
    private final long sequenceNumber;
    // 任务执行的具体时间
    private long time;
    // 任务执行的间隔周期
    private final long period;

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

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

DelayQueue.take()的执行过程

1.  获取Lock
2.  获取周期任务  
    2.1. 如果PriorityQueue为空，**当前线程**到Condition中等待  
    2.2. 如果PriorityQueue的头元素的time时间比当前时间大，到Condition中等待到time时间  
    2.3. 获取PriorityQueue的头元素；如果PriorityQueue不为空，则唤醒在Condition中等待的所有线程
3.  结束释放Lock

ScheduledThreadPoolExecutor在一个循环中执行步骤2，直到线程从PriorityQueue获取到一 个元素之后（执行2.3.1之后），才会退出无限循环（结束步骤2）

# 3.FutureTask #

FutureTask代表异步计算的结果  
FutureTask除了实现Future接口外，还实现了Runnable接口。因此，FutureTask可以交给 Executor执行，也可以由调用线程直接执行（FutureTask.run()）。根据FutureTask.run()方法被执行 的时机，FutureTask可以处于下面3种状态。

*  **未启动**。FutureTask.run()方法还没有被执行之前，FutureTask处于未启动状态。当创建一 个FutureTask，且没有执行FutureTask.run()方法之前，这个FutureTask处于未启动状态。
 *  **已启动**。FutureTask.run()方法被执行的过程中，FutureTask处于已启动状态。
 *  **已完成**。FutureTask.run()方法执行完后正常结束，或被取消（FutureTask.cancel（…）），或 执行FutureTask.run()方法时抛出异常而异常结束，FutureTask处于已完成状态。  
    \[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-RwCukvov-1626072018933)(evernotecid://962030F2-9E96-46F7-8A6E-170AEA738518/appyinxiangcom/23733824/ENResource/p1048)\]  
    可以把FutureTask交给Executor执行；也可以通过ExecutorService.submit（…）方法返回一个 FutureTask，然后执行FutureTask.get()方法或FutureTask.cancel（…）方法。除此以外，还可以单独 使用FutureTask。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzY18xMjM0NTY_size_16_color_FFFFFF_t_70]: /images/20221004/3850a9d5db83440e9493165d4088153d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzY18xMjM0NTY_size_16_color_FFFFFF_t_70 1]: /images/20221004/b878ca2182ab408ba566b3c235d723e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzY18xMjM0NTY_size_16_color_FFFFFF_t_70 2]: /images/20221004/9a1bcf2de6c14b67bac2d3e6cf3f1ffb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzY18xMjM0NTY_size_16_color_FFFFFF_t_70 3]: /images/20221004/0e78ee95c5c04b84afe75172c310d807.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzY18xMjM0NTY_size_16_color_FFFFFF_t_70 4]: /images/20221004/e6f244327f2f43c28efb54e3a5af0c15.png