Java并发框架:Executor

介绍

随着当今处理器中可用的核心数量的增长， 随着对实现更高吞吐量的需求的不断增加，多线程 API 变得很是流行。 Java 提供了本身的多线程框架，称为 Executor 框架.

1. Executor 框架是什么？

Executor 框架包含一组用于有效管理工做线程的组件。Executor API 经过 Executors 将任务的执行与要执行的实际任务解耦。 这是 生产者-消费者 模式的一种实现。

java.util.concurrent.Executors 提供了用于建立工做线程的线程池的工厂方法。

为了使用 Executor 框架，咱们须要建立一个线程池并提交任务给它以供执行。Executor 框架的工做是调度和执行已提交的任务并从线程池中拿到返回的结果。

浮现于脑海中的一个基本的问题是，当咱们建立 java.lang.Thread 的对象或调用实现了 Runnable/Callable 接口来达到的程序的并行性时，为何须要线程池？

答案来源于两个基本面：

1. 为新任务建立新的线程会存在额外的线程建立以及销毁的开销。管理这些线程的生命周期会明显增长 CPU 的执行时间。
2. 不进行任何限制地为每一个进程建立线程会致使建立大量线程。这些线程会占用大量内存并引发资源的浪费。当一个线程利用完 CPU 的时间片后另外一个线程即将利用CPU的时间片时，CPU 会花费大量的时间来切换线程的上下文。

全部的这些因素都会致使系统的吞吐量降低。线程池经过保持线程一直存活并重用这些线程来克服这个问题。当提交到线程池中的任务多于正在执行的线程时，那些多余的任务将被放到队列中。 一旦执行任务的线程有空闲的了，它们会从队列中取下一个任务来执行。对于 JDK 提供的现成的 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 返回的 java.util.concurrent.Future 对象访问提交给 executor 的任务的结果。 Future 能够被认为是 executor 对调用者的响应。

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

如上所述，提交给 executor 的任务是异步的，即程序不会等待当前任务执行完成，而是直接进入下一步。相反，每当任务执行完成时，executor 在此 Future对象中设置它。

调用者能够继续执行主程序，当须要提交任务的结果时，他能够在这个 Future对象上调用.get() 方法来获取。若是任务完成，结果将当即返回给调用者，不然调用者将被阻塞，直到 executor 完成此操做的执行并计算出结果。

若是调用者不能无限期地等待任务执行的结果，那么这个等待时间也能够设置为定时地。能够经过 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，由于这个 demo是在四核处理器上开发的。若是正在执行的任务执行大量 I/O 操做或花费较长时间等待外部资源，则线程数可能超过处理器的核心数。

咱们实例化了 Task 类，并将它提交给 executors 执行。 结果由 Future 对象返回，而后咱们在屏幕上打印。

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

Hello World! 

正如所料，任务追加了问候语 Hello 并经过 Future object 返回结果。

最后，咱们调用 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() 方法不可抛出任何已受检的异常。

5. 总结

随着处理器时钟速度难以提升，多线程正变得愈来愈主流。 可是，因为涉及复杂性，处理每一个线程的生命周期很是困难。

在本文中，咱们展现了一个高效而简单的多线程框架，即 Executor Framework，并解释了它的不一样组件。 咱们还看了一下在 executor 中建立提交和执行任务的不一样示例。

与往常同样，此示例的代码能够在 GitHub上找到。

原文：https://stackabuse.com/concurrency-in-java-the-executor-framework/

做者：Chandan Singh

译者：KeepGoingPawn