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

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. SerialExecutor(Executor executor) {
  6. this.executor = executor;
  7. }
  8. public synchronized void execute(final Runnable r) {
  9. tasks.offer(new Runnable() {
  10. public void run() {
  11. try {
  12. r.run();
  13. } finally {
  14. scheduleNext();
  15. }
  16. }
  17. });
  18. if (active == null) {
  19. scheduleNext();
  20. }
  21. }
  22. protected synchronized void scheduleNext() {
  23. if ((active = tasks.poll()) != null) {
  24. executor.execute(active);
  25. }
  26. }
  27. }

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. executorService.execute(new Runnable() {
  3. public void run() {
  4. System.out.println("Asynchronous task");
  5. }
  6. });
  7. executorService.shutdown();

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

4、任务的委托

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


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

5、ExecutorService 的实现

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

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. executorService.execute(new Runnable() {
  3. public void run() {
  4. System.out.println("Asynchronous task");
  5. }
  6. });
  7. 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. 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. Set<Callable<String>> callables = new HashSet<Callable<String>>();
  3. callables.add(new Callable<String>() {
  4. public String call() throws Exception {
  5. return "Task 1";
  6. }
  7. });
  8. callables.add(new Callable<String>() {
  9. public String call() throws Exception {
  10. return "Task 2";
  11. }
  12. });
  13. callables.add(new Callable<String>() {
  14. public String call() throws Exception {
  15. return "Task 3";
  16. }
  17. });
  18. String result = executorService.invokeAny(callables);
  19. System.out.println("result = " + result);
  20. 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. try {
  12. System.out.println("blocking here");
  13. Integer result = futureTask.get();
  14. System.out.println(result);
  15. } catch (InterruptedException ignored) {
  16. } catch (ExecutionException ignored) {
  17. }

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

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

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

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

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

时间: 01-20

Java并发编程 - Executor,Executors,ExecutorService, CompletionServie,Future,Callable的相关文章

Java并发编程——Executor接口及线程池的使用

在如今的程序里,单线程的程序,应该已经比较少了,而Java语言是内置支持多线程并发的,大家都说Java语言内置支持多线程,非常非常的强大和方便,但一直没有深入研究jdk内concurrent包.今天就认真学习了一下java.util.concurrent包,发现jdk多线程编程果然是强大和方便.本文是学习java.util.concurrent包内线程池及相关接口的一些总结. 任务接口抽象 Runnable接口 在java.lang包内,为多线程提供了Runnable接口. public int

Java并发编程-Executor框架之Callable和Future接口

在上一篇文章中我们已经了解了Executor框架进行线程管理,这篇文章将学习Executor框架的另一个特性,我们知道执行Runnable任务是没有返回值得,但Executor可以运行并发任务并获得返回值,Concurrent包提供下面两个接口实现这个功能: Callable接口:这个接口声明call(),类似于Runnable的run(),可以在这个方法里实现任务的具体逻辑操作.Callable是一个泛型接口,必须声明call()的返回类型. Future接口:这个接口声明了一下方法来获取Ca

Java 并发编程 Executor

Executor框架是指java 5中引入的一系列并发库中与executor相关的一些功能类,其中包括线程池,Executor,Executors,ExecutorService,CompletionService,Future,Callable等.他们的关系为: 并发编程的一种编程方式是把任务拆分为一些列的小任务,即Runnable,然后在提交给一个Executor执行,Executor.execute(Runnalbe) .Executor在执行时使用内部的线程池完成操作. 一.创建线程池

Java并发编程-扩展可回调的Future

前提 最近在看JUC线程池java.util.concurrent.ThreadPoolExecutor的源码实现,其中了解到java.util.concurrent.Future的实现原理.从目前java.util.concurrent.Future的实现来看,虽然实现了异步提交任务,但是任务结果的获取过程需要主动调用Future#get()或者Future#get(long timeout, TimeUnit unit),而前者是阻塞的,后者在异步任务执行时间不确定的情况下有可能需要进行轮询

Java并发编程-Executor框架(转)

本文转自http://blog.csdn.net/chenchaofuck1/article/details/51606224 感谢作者 我们在传统多线程编程创建线程时,常常是创建一些Runnable对象,然后创建对应的Thread对象执行它们,但是如果程序需要并发执行大量的任务时,需要为每个任务都创建一个Thread,进行管理,这将会影响程序的执行效率,并且创建线程过多将会使系统负载过重. 在JDK 1.5之后通过了一套Executor框架能够解决这些问题,能够分解任务的创建和执行过程.该框架

Java并发编程 - Executor框架(一)Executor,

1.并发编程的一种编程方式是把任务拆分为一些列的小任务,即Runnable,然后将这些任务提交给一个Executor执行, Executor.execute(Runnalbe) .Executor在执行时使用其内部的线程池来完成操作. Executor的子接口有:ExecutorService,ScheduledExecutorService,已知实现类:AbstractExecutorService,ScheduledThreadPoolExecutor,ThreadPoolExecutor.

Java并发编程(十四):Callable、Future和FutureTask(转载)

本文转载自:ttp://www.cnblogs.com/dolphin0520/p/3949310.html 在前面的文章中我们讲述了创建线程的2种方式,一种是直接继承Thread,另外一种就是实现Runnable接口. 这2种方式都有一个缺陷就是:在执行完任务之后无法获取执行结果. 如果需要获取执行结果,就必须通过共享变量或者使用线程通信的方式来达到效果,这样使用起来就比较麻烦. 而自从Java 1.5开始,就提供了Callable和Future,通过它们可以在任务执行完毕之后得到任务执行结果

Java并发编程-Executor框架集

Executor框架集对线程调度进行了封装,将任务提交和任务执行解耦. 它提供了线程生命周期调度的所有方法,大大简化了线程调度和同步的门槛. Executor框架集的核心类图如下: 从上往下,可以很清晰的看出框架集的各个类,以及它们之间的关系:Executor,是一个可以提交可执行(Runnable)任务的Object,这个接口解耦了任务提交和执行细节(线程使用.调度等),Executor主要用来替代显示的创建和运行线程:ExecutorService提供了异步的管理一个或多个线程终止.执行过程

java并发编程-Executor框架

一 简介 线程的使用在java中占有极其重要的地位,在jdk1.4极其之前的jdk版本中,关于线程池的使用是极其简陋的.在jdk1.5之后这一情况有了很大的改观.Jdk1.5之后加入了java.util.concurrent包,这个包中主要介绍java中线程以及线程池的使用.为我们在开发中处理线程的问题提供了非常大的帮助. 二:线程池 线程池的作用: 线程池作用就是限制系统中执行线程的数量.     根据系统的环境情况,可以自动或手动设置线程数量,达到运行的最佳效果:少了浪费了系统资源,多了造成