Java中的并发Executor框架

男娘i 2021-12-16 14:53 239阅读 0赞

## 介绍 ##

随着处理器核数的增加，随着对实现更高吞吐量的不断增长的需求，多线程`API`变得非常流行,`Java`提供了它自己的多线程框架[Executor Framework][]。

## Executor 框架是什么？ ##

`Executor Framework`包含一组用于有效管理工作线程的组件。`Executor API`将任务的执行与要通过执行程序执行的实际任务分离。此模式是生产者-消费者实现之一。  
`java.util.concurrent.Executors`可以提供工厂方法被使用创建`ThreadPools`的工作线程。

为了使用`Executor`框架，我们需要创建一个这样的线程池，把任务提交到线程池中让排队执行。`Executor`框架的职责就是计划执行提交的任务并且从线程池中返回结果集。

我们想到的一个基本问题是，当我们可以创建`java.lang.Thread`的对象或实现`Runnable / Callable`接口以实现并行时，为什么我们需要这样的线程池？

答案归结为两个基本事实：

*  为新任务创建新线程会导致线程创建和拆除的开销。管理此线程生命周期会显着增加执行时间。
 *  为每个进程添加新线程而不进行任何限制会导致创建大量线程。这些线程占用内存并导致资源浪费。当每个线程被换出并且另一个线程进入执行时，CPU开始花费太多时间来切换上下文。

这些因素都会降低系统的吞吐量。线程池通过保持线程活动并重用线程来克服此问题。流入的任何多余任务比池中的线程可以处理的都保存在队列中。  
一旦任何线程获得空闲，他们就会从这个队列中获取下一个任务。对于JDK提供的现成执行程序，此任务队列基本上是无界的。

## Executors的类型 ##

现在我们对`Executors`有了很好的认识，那么现在让我们来看看不同种类的`Executors`.

*  SingleThreadExecutor

这个线程池只有一个线程。它是以顺序的方式执行任务。当在执行任务的时候如果 一个线程由于异常而终止，一个新的线程就会被创建来替代老的线程，随后的任务都会在新的线程  
中执行。

ExecutorService executorService = Executors.newSingleThreadExecutor()

*  FixedThreadPool(n)

正如命名的含义一样，这个线程池是有一个固定数量的线程。任务会被线程池中的线程执行，如果有很多的任务，他们会被存储在一个`LinkedBlockingQueue`中。  
此数字通常是底层处理器支持的线程总数。

ExecutorService executorService = Executors.newFixedThreadPool(4);

*  CachedThreadPool

该线程池主要用于需要执行大量短期并行任务的地方。与固定线程池不同，此执行程序池的线程数不受限制。如果所有线程都忙于执行某些任务并且新任务到来，则池将创建并向执行程序添加新线程。  
只要其中一个线程变为空闲，它就会占用新任务的执行。如果一个线程保持空闲60秒，它们将被终止并从缓存中删除。

但是，如果管理不正确，或者任务不是短暂的，则线程池将包含大量活动线程。这可能导致资源颠簸并因此导致性能下降。

ExecutorService executorService = Executors.newCachedThreadPool();

*  ScheduledExecutor

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

ScheduledExecutorService scheduledExecService = Executors.newScheduledThreadPool(1);

个任务可以使用`ScheduledExcutor`中的`scheduleAtFixedRate`或者`scheduleWithFixedDelay`  
来实现。

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

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

这两种方法的主要区别在于它们对连续执行预定作业之间的延迟的解释。

`scheduleAtFixedRate`以固定间隔执行任务，而不管前一个任务何时结束。

`scheduleWithFixedDelay`仅在当前任务完成后才开始延迟倒计时。

## 关于Future对象的理解 ##

可以使用执行程序返回的`java.util.concurrent.Future`对象访问提交执行程序的任务的结果。`Future`可以被认为是通过`excutor`的调用的承诺。

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

如上所述，提交给执行程序的任务是异步的，即程序执行不等待任务执行完成以继续下一步。相反，每当任务执行完成时，执行程序都会在此Future对象中设置它。  
调用方可以继续执行主程序，当任务提交的结果被需要的时候，可以调用`Future`的`.get()`方法。如果任务完成，结果将立即返回给调用者，否则调用者将被阻塞，  
直到执行程序完成此操作的执行并计算结果。

如果调用者无法在检索结果之前无限期地等待，则此等待也可以定时。通过`Future.get(long timeout, TimeUnit unit)`来实现，这个方法会抛出一个`TimeoutException`  
异常。如果结果未在规定的时间内返回。调用者可以处理此异常并继续执行该程序。

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

只有在被提交的任务实现了`java.util.concurrent.Callable`接口的时候，调用`Future.get()`会返回结果；如果任务实现的是`Runnable`接口，那么一旦任务完成，  
调用`.get()`方法会返回`null`。

另一个重要的方法是`Future.cancel（boolean mayInterruptIfRunning）`方法。此方法用于取消已提交任务的执行。  
如果任务已在执行，则执行程序将尝试在`mayInterruptIfRunning`标志传递为`true`时中断任务执行。

## 例子：创建并执行一个简单的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`类型。它也被声明为抛出异常。这种向执行程序和执行程序抛出异常的能力将此异常返回给调用者非常重要，  
因为它有助于调用者知道任务执行的状态。

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

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();
        }
    }

这里我们创建了一个`FixedThreadPool`的线程池，里面包含4个线程。这个演示是在四核处理器上开发的。如果正在执行的任务执行大量I / O操作或花费时间等待外部资源，  
则线程数可能超过处理器核心。

我们已经实例化了`Task`类，并将它传递给执行程序以供执行。结果由`Future`对象返回，然后我们在屏幕上打印。

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

Hello World!

正如所料，任务追加问候语“`Hello`”并通过`Future`对象返回结果。

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

`.shutdown()`方法等待执行程序完成当前提交的任务.但是，如果要求是立即关闭执行程序而不等待，那么我们可以使用`.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 + "!");
        }
    }

无法从`run()`方法返回任务执行的结果。因此，我们直接从这里打印。`run()`方法未配置为抛出任何已检查的异常。

## 结论 ##

随着处理器时钟速度难以提高，多线程正变得越来越主流。但是，由于涉及复杂性，处理每个线程的生命周期非常困难。  
在本文中，我们演示了一个高效而简单的多线程框架，即`Executor Framework`，并解释了它的不同组件。  
我们还看了一下在执行程序中创建提交和执行任务的不同示例。与往常一样，[可以在此示例中找到此示例的代码][Link 1]

> 原文：https://stackabuse.com/concurrency-in-java-the-executor-framework/  
> 作者：Chandan Singh  
> 译者：lee  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI5MzQzMjU_size_16_color_FFFFFF_t_70]

[Executor Framework]: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executor.html
[Link 1]: https://github.com/chand31290/codingExamples/tree/master/ExecutorExample
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI5MzQzMjU_size_16_color_FFFFFF_t_70]: /images/20211214/71cf37ef768047d58e8fb2f64b121ef9.png