搞定面试官谈谈你对JDK中Executor的理解？

Posted 2020-11-26 lovebell

前言

随着当今处理器计算能力愈发强大，可用的核心数量越来越多，各个应用对其实现更高吞吐量的需求的不断增长，多线程 API 变得非常流行。在此背景下，Java自JDK1.5 提供了自己的多线程框架，称为 Executor 框架.

1. Executor 框架是什么？

1.1 简介

Java Doc中是这么描述的

An object that executes submitted Runnable tasks. This interface provides a way of decoupling task submission from the mechanics of how each task will be run, including details of thread use, scheduling, etc. An Executor is normally used instead of explicitly creating threads.

执行提交的Runnable任务的对象。这个接口提供了一种将任务提交与如何运行每个任务的机制，包括线程的详细信息使用、调度等。通常使用Executor而不是显式地创建线程。

我们可以这么理解：Executor就是一个线程池框架，在开发中如果需要创建线程可优先考虑使用Executor，无论你需要多线程还是单线程，Executor为你提供了很多其他功能，包括线程状态，生命周期的管理。

Executor 位于java.util.concurrent.Executors ，提供了用于创建工作线程的线程池的工厂方法。它包含一组用于有效管理工作线程的组件。Executor API 通过 Executors 将任务的执行与要执行的实际任务解耦。 这是 生产者-消费者 模式的一种实现。

浮现于脑海中的一个基本的问题是，当我们创建 java.lang.Thread 对象或调用实现了 Runnable/Callable 接口来实现多线程时，为什么需要线程池？

如果我们不采用线程池，为每一个请求都创建一个线程的话：

1. 管理线程的生命周期开销非常高。管理这些线程的生命周期会明显增加 CPU 的执行时间，会消耗大量计算资源。
2. 线程间上下文切换造成大量资源浪费。
3. 程序稳定性会受到影响。我们知道，创建线程的数量存在一个限制，这个限制将随着平台的不同而不同，并且受多个因素制约，包括jvm的启动参数、Thread构造函数中请求的栈大小，以及底层操作的限制等。如果超过了这个限制，那么很可能抛出OutOfMemoryError异常，这对于运行中的应用来说是非常危险的。

所有的这些因素都会导致系统吞吐量下降。线程池通过保持一些存活线程并重用这些线程来克服这个问题。当提交到线程池中的任务多于线程池最大任务数时，那些多余的任务将被放到一个队列中。 一旦正在执行的线程有空闲了，它们会从队列中取下一个任务来执行。JDK 中的 Executors中， 此任务队列是没有长度限制的。

1.2 实现

我们先来看一下Executor的实现关系。

还是蛮好理解的，正如Java优秀框架的一贯设计思路，顶级接口-次级接口-虚拟实现类-实现类。

Executor：执行者，java线程池框架的最上层父接口，地位类似于spring的BeanFactry、集合框架的Collection接口，在Executor这个接口中只有一个execute方法，该方法的作用是向线程池提交任务并执行。

ExecutorService：该接口继承自Executor接口，添加了shutdown、shutdownAll、submit、invokeAll等一系列对线程的操作方法，该接口比较重要，在使用线程池框架的时候，经常用到该接口。

AbstractExecutorService:这是一个抽象类，实现ExecuotrService接口，

ThreadPoolExecutor：这是Java线程池最核心的一个类，该类继承自AbstractExecutorService，主要功能是创建线程池，给任务分配线程资源，执行任务。

ScheduledExecutorSerivce 和 ScheduledThreadPoolExecutor 提供了另一种线程池：延迟执行和周期性执行的线程池。

Executors：这是一个静态工厂类，该类定义了一系列静态工厂方法，通过这些工厂方法可以返回各种不同的线程池。

2. Executors 的类型

现在我们已经了解了 Executors 是什么， 让我们来看看不同类型的 Executors。

2.1 SingleThreadExecutor

此线程池 Executor 只有一个线程。它用于以顺序方式的形式执行任务。如果此线程在执行任务时因异常而挂掉，则会创建一个新线程来替换此线程，后续任务将在新线程中执行。

ExecutorService executorService = Executors.newSingleThreadExecutor()

2.2 FixedThreadPool(n)

顾名思义，它是一个拥有固定数量线程的线程池。提交给 Executor 的任务由固定的 n 个线程执行，如果有更多的任务，它们存储在 LinkedBlockingQueue 里。这个数字 n 通常跟底层处理器支持的线程总数有关。

ExecutorService executorService = Executors.newFixedThreadPool(4);

2.3 CachedThreadPool

该线程池主要用于执行大量短期并行任务的场景。与固定线程池不同，此线程池的线程数不受限制。如果所有的线程都在忙于执行任务并且又有新的任务到来了，这个线程池将创建一个新的线程并将其提交到 Executor。只要其中一个线程变为空闲，它就会执行新的任务。 如果一个线程有 60 秒的时间都是空闲的，它们将被结束生命周期并从缓存中删除。

但是，如果管理得不合理，或者任务不是很短的，则线程池将包含大量的活动线程。这可能导致资源紊乱并因此导致性能下降。

ExecutorService executorService = Executors.newCachedThreadPool();

2.4 ScheduledExecutor

当我们有一个需要定期运行的任务或者我们希望延迟某个任务时，就会使用此类型的 executor。

ScheduledExecutorService scheduledExecService = Executors.newScheduledThreadPool(1);

可以使用 scheduleAtFixedRate 或 scheduleWithFixedDelay 在 ScheduledExecutor 中定期的执行任务。

scheduledExecService.scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
scheduledExecService.scheduleWithFixedDelay(Runnable command, long initialDelay, long period, TimeUnit unit)

这两种方法的主要区别在于它们对连续执行定期任务之间的延迟的应答。

scheduleAtFixedRate：无论前一个任务何时结束，都以固定间隔执行任务。

scheduleWithFixedDelay：只有在当前任务完成后才会启动延迟倒计时。

3. 对 Future 对象的理解

由于提交给Executor 的任务是异步的，需要有一个对象来接收Executor 的处理结果，这个对象就是java.util.concurrent.Future（类似于JS中的Promise）。

应用方式：

Future<String> result = executorService.submit(callableTask);

调用者可以继续执行主程序，当需要提交任务的结果时，他可以在这个 Future对象上调用.get() 方法来获取。如果任务完成，结果将立即返回给调用者，否则调用者将被阻塞，直到 Executor 完成此操作的执行并计算出结果。（了解JS的童鞋此处可以和Promise的then()相类比）。

如果调用者不能无限期地等待任务执行的结果，那么这个等待时间也可以设置为定时地。可以通过 Future.get(long timeout，TimeUnit unit) 方法实现，如果在规定的时间范围内没有返回结果，则抛出 TimeoutException。调用者可以处理此异常并继续执行该程序。

如果在执行任务时出现异常，则对 get 方法的调用将抛出一个ExecutionException。

对于 Future.get()方法返回的结果，一个重要的事情是，只有提交的任务实现了java.util.concurrent.Callable接口时才返回 Future。如果任务实现了Runnable接口，那么一旦任务完成，对 .get() 方法的调用将返回 null。

另一点是 Future.cancel(boolean mayInterruptIfRunning) 方法。此方法用于取消已提交任务的执行。如果任务已在执行，则 Executor 将尝试在mayInterruptIfRunning 标志为 true 时中断任务执行。

4. Example: 创建和执行一个简单的 Executor

我们现在将创建一个任务并尝试在 fixed pool Executor 中执行它：

public class Task implements Callable<String> {

    private String message;

    public Task(String message) {
        this.message = message;
    }

    @Override
    public String call() throws Exception {
        return "Hello " + message + "!";
    }
}

Task 类实现 Callable 接口并有一个 String 类型作为返回值的方法。 这个方法也可以抛出 Exception。这种向 Executor 抛出异常的能力以及 Executor 将此异常返回给调用者的能力非常重要，因为它有助于调用者知道任务执行的状态。

现在让我们来执行一下这个任务：

public class ExecutorExample {  
    public static void main(String[] args) {

        Task task = new Task("World");

        ExecutorService executorService = Executors.newFixedThreadPool(4);
        Future<String> result = executorService.submit(task);

        try {
            System.out.println(result.get());
        } catch (InterruptedException | ExecutionException e) {
            System.out.println("Error occured while executing the submitted task");
            e.printStackTrace();
        }

        executorService.shutdown();
    }
}

我们创建了一个具有4个线程数的 FixedThreadPool Executors，并实例化了 Task 类，并将它提交给 Executors 执行。 结果由 Future 对象返回，然后我们在屏幕上打印。

让我们运行 ExecutorExample 并查看其输出：

Hello World!

最后，我们调用 executorService 对象上的 shutdown 来终止所有线程并将资源返回给 OS。

shutdown() 方法等待 Executor 完成当前提交的任务。 但是，如果要求是立即关闭 Executor 而不等待，那么我们可以使用 shutdownNow() 方法。

任何待执行的任务都将结果返回到 java.util.List 对象中。

我们也可以通过实现 Runnable 接口来创建同样的任务：

public class Task implements Runnable{

    private String message;

    public Task(String message) {
        this.message = message;
    }

    public void run() {
        System.out.println("Hello " + message + "!");
    }
}

当我们实现 Runnable 时，这里有一些重要的变化。

1. 无法从 run() 方法得到任务执行的结果。 因此，我们直接在这里打印。
2. run() 方法不可抛出任何已受检的异常。

Notes：如何合理配置线程池的大小

一般需要根据任务的类型来配置线程池大小：

如果是CPU密集型任务，就需要尽量压榨CPU，参考值可以设为 NCPU+1
如果是IO密集型任务，参考值可以设置为2*NCPU
当然，这只是一个参考值，具体的设置还需要根据实际情况进行调整，比如可以先将线程池大小设置为参考值，再观察任务运行情况和系统负载、资源利用率来进行适当调整。