Java并发编程 - Executor，Executors，ExecutorService, CompletionServie,Future,Callable

一、Exectuor框架简介

Java从1.5版本开始，为简化多线程并发编程，引入全新的并发编程包:java.util.concurrent及其并发编程框架（Executor框架）。 Executor框架是指java 5中引入的一系列并发库中与executor相关的一些功能类，其中包括线程池，Executor，Executors，ExecutorService，CompletionService，Future，Callable等。他们的关系为

在Executor框架中，使用执行器(Exectuor)来管理Thread对象，从而简化了并发编程。

二、认识Exectuor（执行器）

1、并发编程的一种编程方式是把任务拆分为一系列的小任务，即Runnable，然后将这些任务提交给一个Executor执行，Executor.execute(Runnalbe) 。Executor在执行时使用其内部的线程池来完成操作。

Executor的子接口有：ExecutorService,ScheduledExecutorService,已知实现类：AbstractExecutorService,ScheduledThreadPoolExecutor,ThreadPoolExecutor。

2、Executor属于public类型的接口。可以用于提交，管理或者执行Runnable任务。实现Executor接口的class还可以控制Runnable任务执行线程的具体细节。包括线程使用的细节、调度等。一般来说，Runnable任务开辟在新线程中的使用方法为：new Thread(new RunnableTask())).start()

3、但在Executor中，可以使用Executor而不用显示地创建线程。例如，可以使用以下方法创建线程，而不是像第2点中为一种任务中的每个任务都调用new Thread(...)的方法。

Exectuor executor = anExecutor();
executor.execute(new RunnableTask()); // 异步执行
executor.execute(new RunnableTask());

4、但是，Executor接口并没有严格地要求执行必须是异步/同步的，一切都相当自由。在最简单的情况下，执行程序可以在调用者的线程中立即运行已提交的任务，

class DirectExecutor implements Executor {      
       public void execute(Runnable r) {          
              r.run();    
       }  
}

更常见的是，任务在某个不是调用者线程的线程中执行的。如在另一个线程中启动：

class ThreadPerTaskExecutor implements Executor {      
           public void execute(Runnable r) {          
              new Thread(r).start();      
            }  
}

也可以在实现中用另一个Executor来序列化执行过程：

class SerialExecutor implements Executor {  
    final Queue<Runnable> tasks = new ArrayDeque<Runnable>();  
    final Executor executor;  
    Runnable active;  
  
    SerialExecutor(Executor executor) {  
        this.executor = executor;  
    }  
  
    public synchronized void execute(final Runnable r) {  
        tasks.offer(new Runnable() {  
            public void run() {  
                try {  
                    r.run();  
                } finally {  
                    scheduleNext();  
                }  
            }  
        });  
        if (active == null) {  
            scheduleNext();  
        }  
    }  
  
    protected synchronized void scheduleNext() {  
        if ((active = tasks.poll()) != null) {  
            executor.execute(active);  
        }  
    }  
}

5、ThreadPoolExecutor类提供了一个可供可扩展的线程池实现。Executors类为Executor接口及其实现提供了便捷的工厂方法。

6、 Executor中的方法execute。void execute(Runnable command)表示在未来的某个时间执行给定的命令。该命令可能在新的线程、已经入池的线程或者正在调用的线程中执行。

三、Executors类：主要用于提供线程池相关的操作

Executors类，提供了一系列工厂方法用于创建线程池，返回的线程池都实现了ExecutorService接口。

1、public static ExecutorService newFiexedThreadPool(int Threads) 创建固定数目线程的线程池。

2、public static ExecutorService newCachedThreadPool()：创建一个可缓存的线程池，调用execute 将重用以前构造的线程（如果线程可用）。如果没有可用的线程，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。

3、public static ExecutorService newSingleThreadExecutor()：创建一个单线程化的Executor。

4、public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)

创建一个支持定时及周期性的任务执行的线程池，多数情况下可用来替代Timer类。

四、ExecutorService与生命周期

1、ExecutorService可以理解为程序员提供了一堆操作Executor的API

2、ExecutorService扩展了Executor并添加了一些生命周期管理的方法。一个Executor的生命周期有三种状态

运行、关闭和终止。

Executor创建时处于运行状态。当调用ExecutorService.shutdown()后，处于关闭状态，isShutdown()方法返回true。这时，不应该再向Executor中添加任务，所有已添加的任务执行完毕后，Executor处于终止状态，isTerminated()返回true。如果Executor处于关闭状态，往Executor提交任务会抛出unchecked exception RejectedExecutionException。

3、本质

接口ExecutorService 表述了异步执行的机制，并且可以让任务在后台执行。一个ExecutorService 实例因此特别像一个线程池。事实上，在 java.util.concurrent 包中的 ExecutorService 的实现就是一个线程池的实现。

ExecutorService executorService = Executors.newFixedThreadPool(10);
 
executorService.execute(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});
 
executorService.shutdown();

该示例代码首先使用 newFixedThreadPool() 工厂方法创建一个ExecutorService ，上述代码创建了一个可以容纳10个线程任务的线程池。其次，向 execute() 方法中传递一个异步的 Runnable 接口的实现，这样做会让 ExecutorService 中的某个线程执行这个Runnable 线程。

4、任务的委托

下方展示了一个线程的把任务委托异步执行的ExecutorService的示意图。

Java并发编程 - Executor，Executors，ExecutorService, CompletionServie,Future,Callable

一旦线程把任务委托给 ExecutorService，该线程就会继续执行与运行任务无关的其它任务。

5、ExecutorService 的实现

由于 ExecutorService 只是一个接口，ExecutorService 接口在 java.util.concurrent 包中有如下实现类：

ThreadPoolExecutor
ScheduledThreadPoolExecutor

6、ExecutorService 使用方法

这里有几种不同的方式让你将任务委托给一个ExecutorService：

execute(Runnable)
submit(Runnable)
submit(Callable)
invokeAny()
invokeAll()

7、execute(Runnable)

方法 execute(Runnable) 接收一个java.lang.Runnable 对象作为参数，并且以异步的方式执行它。如下是一个使用 ExecutorService 执行 Runnable 的例子：

ExecutorService executorService = Executors.newSingleThreadExecutor();
 
executorService.execute(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});
     
executorService.shutdown();

使用这种方式没有办法获取执行 Runnable 之后的结果，如果你希望获取运行之后的返回值，就必须使用接收 Callable 参数的 execute() 方法。接下来会提到。

8、submit(Runnable)

方法 submit(Runnable) 同样接收一个Runnable 的实现作为参数，但是会返回一个Future 对象。这个Future 对象可以用于判断 Runnable 是否结束执行。如下是一个ExecutorService 的 submit() 方法的例子：

Future future = executorService.submit(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});
//如果任务结束执行则返回 null
System.out.println("future.get()=" + future.get());

9、submit(Callable)

方法 submit(Callable) 和方法 submit(Runnable) 比较类似，但是区别则在于它们接收不同的参数类型。Callable 的实例与 Runnable 的实例很类似，但是 Callable 的 call() 方法可以返回一个结果。方法 Runnable.run() 则不能返回结果。

Callable 的返回值可以从方法 submit(Callable) 返回的 Future 对象中获取。如下是一个 ExecutorService Callable 的样例：

Future future = executorService.submit(new Callable(){
    public Object call() throws Exception {
        System.out.println("Asynchronous Callable");
        return "Callable Result";
    }
});
 
System.out.println("future.get() = " + future.get());

上述样例代码会输出如下结果：

Asynchronous Callable
future.get() = Callable Result

10、inVokeAny()

方法 invokeAny() 接收一个包含 Callable 对象的集合作为参数。调用该方法不会返回 Future 对象，而是返回集合中某一个Callable 对象的结果，而且无法保证调用之后返回的结果是哪一个 Callable，只知道它是这些 Callable 中一个执行结束的 Callable 对象。如果一个任务运行完毕或者抛出异常，方法会取消其它的 Callable 的执行。

以下是一个样例：

ExecutorService executorService = Executors.newSingleThreadExecutor();
 
Set<Callable<String>> callables = new HashSet<Callable<String>>();
 
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 1";
    }
});
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 2";
    }
});
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 3";
    }
});
 
String result = executorService.invokeAny(callables);
 
System.out.println("result = " + result);
 
executorService.shutdown();

以上样例代码会打印出在给定的集合中的某一个Callable 的返回结果。尝试运行后发现每次结果都在改变。有时候返回结果是"Task 1"，有时候是"Task 2"，等等。

11、invokeAll()

方法 invokeAll() 会调用存在于参数集合中的所有 Callable 对象，并且返回一个包含 Future 对象的集合，你可以通过这个返回的集合来管理每个 Callable 的执行结果。需要注意的是，任务有可能因为异常而导致运行结束，所以它可能并不是真的成功运行了。但是我们没有办法通过 Future 对象来了解到这个差异。

12、ExecutorService服务的关闭

当使用 ExecutorService 完毕之后，我们应该关闭它，这样才能保证线程不会继续保持运行状态。

举例来说，如果你的程序通过 main() 方法启动，并且主线程退出了你的程序，如果还有一个活动的 ExecutorService 存在于程序中，那么程序将会继续保持运行状态。存在于 ExecutorService 中的活动线程会阻止Java虚拟机关闭。

为了关闭在 ExecutorService 中的线程，需要调用 shutdown() 方法。但ExecutorService 并不会马上关闭，而是不再接收新的任务，一旦所有的线程结束执行当前任务，ExecutorServie 才会真的关闭。所有在调用 shutdown() 方法之前提交到 ExecutorService 的任务都会执行。

如果你希望立即关闭 ExecutorService，你可以调用 shutdownNow() 方法。这个方法会尝试马上关闭所有正在执行的任务，并且跳过所有已经提交但是还没有运行的任务。但是对于正在执行的任务，是否能够成功关闭它是无法保证的，有可能他们真的被关闭掉了，也有可能它会一直执行到任务结束。这是一个最好的尝试。

五、CompletionService

根据上面的介绍我们知道，现在在Java中使用多线程通常不会再使用Thread对象了。而是会用到java.util.concurrent包下的ExecutorService来初始化一个线程池供我们使用。使用ExecutorService类的时候，我们常维护一个list保存submit的callable task所返回的Future对象。然后在主线程中遍历这个list并调用Future的get()方法取到Task的返回值。

其实除了使用ExecutorService外，还可通过CompletionService包装ExecutorService，然后调用其take()方法去取Future对象。

CompletionService和ExecutorService的主要的区别在于submit的task不一定是按照加入自己维护的list顺序完成的。

ExecutorService中从list中遍历的每个Future对象并不一定处于完成状态，这时调用get()方法就会被阻塞住，如果系统是设计成每个线程完成后就能根据其结果继续做后面的事，这样对于处于list后面的但是先完成的线程就会增加了额外的等待时间。

而CompletionService的实现是维护一个保存Future对象的BlockingQueue。只有当这个Future对象状态是结束的时候，才会加入到这个Queue中，take()方法其实就是Producer-Consumer中的Consumer。它会从Queue中取出Future对象，如果Queue是空的，就会阻塞在那里，直到有完成的Future对象加入到Queue中。所以，先完成的必定先被取出。这样就减少了不必要的等待时间。

六、使用Callable，Future返回结果

Future<V>代表一个异步执行的操作，通过get()方法可以获得操作的结果，如果异步操作还没有完成，则，get()会使当前线程阻塞。FutureTask<V>实现了Future<V>和Runable<V>。Callable代表一个有返回值的操作。

Callable<Integer> func = new Callable<Integer>(){
public Integer call() throws Exception {
System.out.println("inside callable");
Thread.sleep(1000);
return new Integer(8);
}
};
FutureTask<Integer> futureTask = new FutureTask<Integer>(func);
Thread newThread = new Thread(futureTask);
newThread.start();
try {
System.out.println("blocking here");
Integer result = futureTask.get();
System.out.println(result);
} catch (InterruptedException ignored) {
} catch (ExecutionException ignored) {
}

ExecutoreService提供了submit()方法，传递一个Callable，或Runnable，返回Future。如果Executor后台线程池还没有完成Callable的计算，则调用返回Future对象的get()方法，会阻塞直到计算完成。

Java5以后可以利用Future来跟踪异步计算的结果。在此之前主线程要想获得工作线程（异步计算线程）的结果是比较麻烦的事情，需要我们进行特殊的程序结构设计，比较繁琐而且容易出错。有了Future我们就可以设计出比较优雅的异步计算程序结构模型：根据分而治之的思想，我们可以把异步计算的线程按照职责分为3类：

1. 异步计算的发起线程（控制线程）：负责异步计算任务的分解和发起，把分解好的任务交给异步计算的work线程去执行，发起异步计算后，发起线程可以获得Futrue的集合，从而可以跟踪异步计算结果。

2. 异步计算work线程：负责具体的计算任务

3. 异步计算结果收集线程：从发起线程那里获得Future的集合，并负责监控Future的状态，根据Future的状态来处理异步计算的结果。

Java并发编程 - Executor，Executors，ExecutorService, CompletionServie,Future,Callable

继续阅读

多线程下嵌套异步任务导致程序假死

CompletionService 源码解析

WAIC 2023: AI, LLMs, and more hopes for the future

线程池ExecutorService、CompletableFuture异步方法解析

CompletableFuture案例详解

java 利用Future做超时任务处理 java 利用Future做超时任务处理

黑马程序员_java并发Executors

并发编程-FutureTask解析 | 京东物流技术团队

并发编程-FutureTask解析

认识Thread读书笔记

Future丨量子计算机的里程碑时刻，《流浪地球2》MOSS照进现实

Java多线程代码实例总结Java多线程代码实例总结

JAVA并发-Future

concurrent.futures

java 多线程学习-----初入线程池