Executor框架完整解读

Posted 2021-03-03 gocode

1 前言

Java的线程既是工作单元，也是执行机制。从JDK 5开始，把工作单元与执行机制分离开来。工作单元包括Runnable和Callable，而执行机制由Executor框架提供.

在HotSpot VM的线程模型中，Java线程被一对一映射为本地操作系统线程。Java线程启动时会创建一个本地操作系统线程；当该Java线程终止时，这个操作系统线程也会被回收。操作系统会调度所有线程并将它们分配给可用的CPU。在上层，Java多线程程序通常把应用分解为若干个任务，然后使用用户级的调度器（Executor框架）将这些任务映射为固定数量的线程；在底层，操作系统内核将这些线程映射到硬件处理器上。应用程序通过Executor框架控制上层的调度；而下层的调度由操作系统内核控制，下层的调度不受应用程序的控制。

2 Executor框架组成

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

• ①任务： Runnable或Callable接口及其实现类

• ②任务执行器： 主要是Executor及扩展Executor的ExecutorService接口的一些实现类。Executor框架有两个重要的实现类，一个是线程池执行器ThreadPoolExecutor、另一个是定时任务执行器ScheduledThreadPoolExecutor .

• ③任务的结果: Future接口及其默认实现FutureTask

说明：

Runnable接口(无返回值)和Callable接口(有返回值)的实现类，都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor执行。

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

3 Runnable和Callable

Runnable接口和Callable接口的实现类，都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor执行。它们之间的区别是Runnable没有返回值，无法判断任务是否完成，而Callable有一个返回结果。

除了可以自己创建实现Callable接口的对象外，还可以使用工厂类Executors将一个Runnable包装成一个Callable.

callable(Runnable )方法包装不需要结果的Runnable任务，任务完成后Future.get()会获取一个null值；而callable(Runnable,T)可以包装一个需要结果的Runnable任务，结果可以通过参数result指定，任务完成后Future.get()可以获得这个结果result.

    public static Callable<Object> callable(Runnable task) 
    public static <T> Callable<T> callable(Runnable task, T result)

 

4 ThreadPoolExecutor

ThreadPoolExecutor是ExecutorService的最重要的实现类，ThreadPoolExecutor不直接实现ExecutorService接口，它直接继承于AbstractExecutorService抽象类， AbstractExecutorService对ExecutorSerivice接口中的一些方法做过的默认实现 。

 

 之前的文章线程池ThreadPoolExecutor简介对使用构造方法创建线程池已做详细说明，这里介绍使用工厂类Executors来创建线程池。ThreadPoolExecutor可以使用工厂类Executors提供了一些静态工厂方法，可以以此方便地创建一些常用配置的线程池。Executors可以创建3种类型的ThreadPoolExecutor .

1) 固定线程池

newFixedThreadPool系列方法创建固定线程数的线程池。它适用于为了满足资源管理的需求，而需要限制当前线程数量的应用场景，它适用于负载比较重的服务器。

    //创建固定线程数的线程池
 public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>());
  }
 public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(), threadFactory);
  }

这种线程池的corePoolSize和maximumPoolSize都被设置为参数nThreads。当线程池中的线程数大于corePoolSize时，keepAliveTime为非核心线程等待新任务的最长时间，超过这个时间后的这些线程将被终止。corePoolSize和maximumPoolSize参数设置为相同值,表示线程池中所有线程均是核心线程, 那么keepAliveTime参数就是无意义的.

 

 处理任务流程说明：

①当线程池中的线程数小于corePoolSize时，线程池会创建新线程去执行任务。

②在经过一段时间预热后，线程数达到了corePoolSize（因为maximumPoolSize与corePoolSize相同，此时也达到了最大线程数，以后不会再创建线程），开始将任务放入工作队列中。

③此后有新任务到达就向工作队列中放入（LinkedBlockingQueue无参构造方法，创建的队列容量是Integer.MAX_VALUE ,这种队列几乎不可能容量爆满，不会拒绝任务，拒绝策略不起作用），若有线程处于空闲状态则从工作队列中获取任务并执行。

2) 单线程池

newSingleThreadExecutor系列方法创建单个线程的线程池 它适用于保证任务按顺序执行，并且在任何时候最多只有一个活动(正在执行)的任务。

  //创建单个线程的线程池
  public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
    return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,  new LinkedBlockingQueue<Runnable>(),threadFactory));
  }
  public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,  0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()));
  }

newSingleThreadExecutor方法不是直接返回ThreadPoolExecutor对象，它将ThreadPoolExecutor对象进行包装成FinalizableDelegatedExecutorService对象，但实际的业务处理还是委托给ThreadPoolExecutor去实现。

这类线程池的corePoolSize和maximumPoolSize都被设为1，线程池中最多有一个线程，同样地这里的keepAliveTime参数是无意义的，它使用一个无限大容量的阻塞队列作为存放任务的容器。

 

 处理任务流程说明：

①当线程池中的无任何线程时，线程池会创建一个线程去执行任务。

②当线程池中有一个线程时，开始将任务放入工作队列中。

③此后有新任务到达就向工作队列中放入（LinkedBlockingQueue无参构造方法，创建的队列容量是Integer.MAX_VALUE ,这种队列几乎不可能容量爆满，不会拒绝任务，拒绝策略不起作用），若线程池中的唯一线程处于空闲状态则从工作队列中取出任务并执行。

3) 缓存线程池

newCachedThreadPool系列方法会根据需要创建新线程的线程池(但若之前创建的线程可用，则将复用这些线程) . 它对线程数没有限制，会按需创建新线程，适用于执行很多的短期异步任务的小任务，或者是负载较轻的服务器。

    //根据需要创建新线程的线程池
  public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>(),threadFactory);
  }
  public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());
  }

这种线程池的corePoolSize为零、maximumPoolSize为Integer.MAX_VALUE，表明线程池中没有核心线程、所有线程均是非核心线程，且可允许的最大线程数是近乎无限大。keepAliveTime参数设为60,表明空闲线程最多等待60秒就被终止。

这里使用SynchronousQueue作为工作队列，这种队列是没有容量的（当尝试排队时，只有正好有空闲线程正在等待接受任务时才会入队成功），但可允许创建的最大线程数是无限大的。这意味着主线程提交任务的速度大于任务处理的速度，线程池就会创不断建新线程，这样可能导致创建的线程过多，系统资源被耗尽、程序崩溃。

 

 处理任务流程说明：

①因为核心线程数为0，所以线程池在启动时核心线程池就已经满了。在主线程提交第一个任务时，线程池就要将尝试此任务入队，由于SynchronousQueue的特殊性，只有当此时空闲线程也正在出队,入队与出队两者恰好匹配时，主线程会把任务交给空闲线程去执行。否则将进入下一步。

②当线程池中无任何线程或无空闲线程时，将没有线程执行出队操作。此时线程池会创建建一个新线程执行任务

③上一步中创建的新线程在执行完的任务后，会调用SynchronousQueue.poll等待任务出队。这个空闲线程最多等待60秒时间，若主线程在60秒内提交了一个新任务，此空闲线程将获取到这个任务并执行。若等待60秒后，还没等到新任务到达，这个线程将被终止。

5 ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor是定时任务执行器，它可以通过构造方法创建，也可通过工厂类Executors的静态方法创建。

ScheduledThreadPoolExecutor的构造方法逻辑十分简单，它们直接调用父类实现。

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

 

Executors可创建两种类型的ScheduledThreadPoolExecutor。

①newSingleThreadScheduledExecutor系列方法用于创建单个线程的定时任务执行器。它适用于按照固定顺序执行周期性的定时任务，且最多同时执行一个任务的情况。

②newScheduledThreadPool系列方法用于创建给定个数线程的定时任务执行器。它适用于需要多个线程执行周期任务，同时又要限制线程数、防止创建线程过多耗费资源的情况。

   //单线程的定时任务执行器
     public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
        return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1, threadFactory));
    }
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }
    //多线程的定时任务执行器
     public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }
    public static ScheduledExecutorService newScheduledThreadPool( int corePoolSize, ThreadFactory threadFactory) {
        return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
    }

 

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，并实现了表示定时执行器的接口ScheduledExecutorService。它主要用来在给定的延迟之后运行任务，或者定期执行任务.

 

 DelayedWorkQueue是ScheduledThreadPoolExecutor的一个静态内部类，它是一个无界队列，所以父类ThreadPoolExecutor中的maximumPoolSize、keepAliveTime这两个参数无意义、没有任何效果。

执行任务基本流程：

①当调用ScheduledThreadPoolExecutor的scheduleAtFixedRate()方法或者scheduleWithFixedDelay()方法时，会向的DelayedWorkQueue添加一个实现了RunnableScheduledFuture接口的ScheduledFutureTask类型任务。

②线程池中的线程从DelayedWorkQueue中尝试获取到期任务，若没有任务到期此线程将阻塞等待，直到真正获取到一个任务，然后执行此任务。

 

 ScheduledThreadPoolExecutor会把待调度的任务ScheduledFutureTask放到一个DelayedWorkQueue中，我们来了解一下ScheduledFutureTask。

 ScheduledFutureTask是ScheduledThreadPoolExecutor的一个成员内部类,它继承了FutureTask类，另外还实现了表示周期性任务的ScheduledFutureTask接口，ScheduledFutureTask有3 个重要的成员变量。

• long型成员变量time，表示这个任务将要被执行的具体时间。

• long型成员变量sequenceNumber，表示这个任务被添加到ScheduledThreadPoolExecutor中的序号。

• long型成员变量period，表示任务执行的间隔周期

DelayedWorkQueue封装了一个RunnableScheduledFuture数组，它利用这个数组实现一个基于堆排序的优先级队列，其原理与PriorityQueue类似。ScheduledFutureTask任务会放入这个数组中，DelayedWorkQueue会对数组中的任务按照优先级排序。排序的基本原则是： time小的排在前面，如果两个任务的time相同，就比较sequenceNumber，sequenceNumber小的排在前面。换句话说，时间早的任务先执行，若几个任务同样早，就看谁先提交，先提交的任务先执行。

  执行一个任务的完整步骤：

①某空闲线程从DelayedWorkQueue中获取已到期的ScheduledFutureTask任务(到期任务是指ScheduledFutureTask的time大于等于当前时间)。

②此线程执行这个ScheduledFutureTask任务。

③此线程修改ScheduledFutureTask的time变量为下次将要被执行的时间。

④此线程把这个time被修改后的ScheduledFutureTask重新放回DelayedWorkQueue中。

6 Future

Future接口和FutureTask类用来表示异步任务的结果。当我们把Runnable接口或Callable接口的实现类提交(使用ExecutorService的submit系列方法提交任务)给ThreadPoolExecutor或ScheduledThreadPoolExecutor时，它们将返回一个Future类型的对象(实际返回FutureTask类型对象) , 在调用Future.get()方法时会阻塞等待任务完成后再返回任务的结果。

submit方法提交任务

    public Future<?> submit(Runnable task) 
    public <T> Future<T> submit(Runnable task, T result)
    public <T> Future<T> submit(Callable<T> task)

 

get方法等待任务执行完成并获取结果

 public V get() throws InterruptedException, ExecutionException
 public V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException

 

Future、FutureTask的详细分析在之前的文章FutureTask源码完整解读已有说明，这里不再赘述。

参考：《Java并发编程的艺术》、《Java的逻辑》