首页 > 代码库 > Java并发编程 - Executor，Executors，ExecutorService, CompletionServie,Future,Callable

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

2024-09-01 20:44:17   218人阅读

一、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(...)的方法。

 

Java代码  技术分享
  1. Exectuor executor = anExecutor();  
  2. executor.execute(new RunnableTask()); // 异步执行  
  3. executor.execute(new RunnableTask());  

 

 

 

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

 

Java代码  技术分享
  1. class DirectExecutor implements Executor {        
  2.        public void execute(Runnable r) {            
  3.               r.run();      
  4.        }    
  5. }  

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

 

Java代码  技术分享
  1. class ThreadPerTaskExecutor implements Executor {        
  2.            public void execute(Runnable r) {            
  3.               new Thread(r).start();        
  4.             }    
  5. }  

 

 

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

 

Java代码  技术分享
  1. class SerialExecutor implements Executor {    
  2.     final Queue<Runnable> tasks = new ArrayDeque<Runnable>();    
  3.     final Executor executor;    
  4.     Runnable active;    
  5.     
  6.     SerialExecutor(Executor executor) {    
  7.         this.executor = executor;    
  8.     }    
  9.     
  10.     public synchronized void execute(final Runnable r) {    
  11.         tasks.offer(new Runnable() {    
  12.             public void run() {    
  13.                 try {    
  14.                     r.run();    
  15.                 } finally {    
  16.                     scheduleNext();    
  17.                 }    
  18.             }    
  19.         });    
  20.         if (active == null) {    
  21.             scheduleNext();    
  22.         }    
  23.     }    
  24.     
  25.     protected synchronized void scheduleNext() {    
  26.         if ((active = tasks.poll()) != null) {    
  27.             executor.execute(active);    
  28.         }    
  29.     }    
  30. }    

 

 

 

 

 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 的实现就是一个线程池的实现。

 

Java代码  技术分享
  1. ExecutorService executorService = Executors.newFixedThreadPool(10);  
  2.    
  3. executorService.execute(new Runnable() {  
  4.     public void run() {  
  5.         System.out.println("Asynchronous task");  
  6.     }  
  7. });  
  8.    
  9. executorService.shutdown();  

 

 

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

 

4、任务的委托

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


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

 

5、ExecutorService 的实现

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

  • ThreadPoolExecutor

  • ScheduledThreadPoolExecutor

6、ExecutorService 使用方法

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

 

Java代码  技术分享
  1. execute(Runnable)  
  2. submit(Runnable)  
  3. submit(Callable)  
  4. invokeAny()  
  5. invokeAll()  

 

 

7、execute(Runnable)

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

 

Java代码  技术分享
  1. ExecutorService executorService = Executors.newSingleThreadExecutor();  
  2.    
  3. executorService.execute(new Runnable() {  
  4.     public void run() {  
  5.         System.out.println("Asynchronous task");  
  6.     }  
  7. });  
  8.        
  9. executorService.shutdown();  

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

 

 

8、submit(Runnable)

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

 

Java代码  技术分享
  1. Future future = executorService.submit(new Runnable() {  
  2.     public void run() {  
  3.         System.out.println("Asynchronous task");  
  4.     }  
  5. });  
  6. //如果任务结束执行则返回 null  
  7. 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 的样例：

 

Java代码  技术分享
  1. Future future = executorService.submit(new Callable(){  
  2.     public Object call() throws Exception {  
  3.         System.out.println("Asynchronous Callable");  
  4.         return "Callable Result";  
  5.     }  
  6. });  
  7.    
  8. System.out.println("future.get() = " + future.get());  

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

 

Java代码  技术分享
  1. Asynchronous Callable  
  2. future.get() = Callable Result  

 

10、inVokeAny()

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

Java代码  技术分享
  1. ExecutorService executorService = Executors.newSingleThreadExecutor();  
  2.    
  3. Set<Callable<String>> callables = new HashSet<Callable<String>>();  
  4.    
  5. callables.add(new Callable<String>() {  
  6.     public String call() throws Exception {  
  7.         return "Task 1";  
  8.     }  
  9. });  
  10. callables.add(new Callable<String>() {  
  11.     public String call() throws Exception {  
  12.         return "Task 2";  
  13.     }  
  14. });  
  15. callables.add(new Callable<String>() {  
  16.     public String call() throws Exception {  
  17.         return "Task 3";  
  18.     }  
  19. });  
  20.    
  21. String result = executorService.invokeAny(callables);  
  22.    
  23. System.out.println("result = " + result);  
  24.    
  25. 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代表一个有返回值的操作。

  1. Callable<Integer> func = new Callable<Integer>(){  
  2.     public Integer call() throws Exception {  
  3.         System.out.println("inside callable");  
  4.         Thread.sleep(1000);  
  5.         return new Integer(8);  
  6.     }         
  7. };        
  8. FutureTask<Integer> futureTask  = new FutureTask<Integer>(func);  
  9. Thread newThread = new Thread(futureTask);  
  10. newThread.start();  
  11.   
  12. try {  
  13.     System.out.println("blocking here");  
  14.     Integer result = futureTask.get();  
  15.     System.out.println(result);  
  16. } catch (InterruptedException ignored) {  
  17. } catch (ExecutionException ignored) {  
  18. } 

 

       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


联系
我们