首页 > 代码库 > 多线程(二)

多线程(二)

看到写的比较详细的一片说明invokeAll 和 invokeAny 的方法, 修改一下,拿来说明问题。

ExecutorService是JDK并发工具包提供的一个核心接口,相当于一个线程池,提供执行任务和管理生命周期的方法。ExecutorService接口中的大部分API都是比较容易上手使用的,本文主要介绍下invokeAll和invokeAny方法的特性和使用。我们先提供几个任务类:一个耗时任务,一个异常任务,一个短时任务。他们会在接下来的测试代码中使用。

  1. package tasks;  
  2. import java.util.concurrent.Callable;  
  3. import java.util.concurrent.TimeUnit;  
  4. public class SleepSecondsCallable implements Callable<String>  
  5. {  
  6. private String name;  
  7. private int seconds;  
  8. public SleepSecondsCallable(String name, int seconds)  
  9.     {  
  10. this.name = name;  
  11. this.seconds = seconds;  
  12.     }  
  13. public String call() throws Exception  
  14.     {  
  15.         System.out.println(name + ",begin to execute");  
  16. try
  17.         {  
  18.             TimeUnit.SECONDS.sleep(seconds);  
  19.         } catch (InterruptedException e)  
  20.         {  
  21.             System.out.println(name + " was disturbed during sleeping.");  
  22.             e.printStackTrace();  
  23. return name + "_SleepSecondsCallable_failed";  
  24.         }  
  25.         System.out.println(name + ",success to execute");  
  26. return name + "_SleepSecondsCallable_succes";  
  27.     }  

这是一个通过睡眠来模拟的耗时任务,该任务是可中断/可终止的任务,能够响应中断请求。

[java] view plaincopy

  1. package tasks;  
  2. import java.util.concurrent.Callable;  
  3. public class ExceptionCallable implements Callable<String>  
  4. {  
  5. private String name = null;  
  6. public ExceptionCallable()  
  7.     {  
  8.     }  
  9. public ExceptionCallable(String name)  
  10.     {  
  11. this.name = name;  
  12.     }  
  13. @Override
  14. public String call() throws Exception  
  15.     {  
  16.         System.out.println("begin to ExceptionCallable.");  
  17.         System.out.println(name.length());  
  18.         System.out.println("end to ExceptionCallable.");  
  19. return name;  
  20.     }  

这是一个可能会在执行过程中,抛出空指针异常的任务。

[java] view plaincopy

  1. package tasks;  
  2. import java.util.Random;  
  3. import java.util.concurrent.Callable;  
  4. public class RandomTenCharsTask implements Callable<String>  
  5. {  
  6. @Override
  7. public String call() throws Exception  
  8.     {  
  9.         System.out.println("RandomTenCharsTask begin to execute...");  
  10.         StringBuffer content = new StringBuffer();  
  11.         String base = "abcdefghijklmnopqrstuvwxyz0123456789";  
  12.         Random random = new Random();  
  13. for (int i = 0; i < 10; i++)  
  14.         {  
  15. int number = random.nextInt(base.length());  
  16.             content.append(base.charAt(number));  
  17.         }  
  18.         System.out.println("RandomTenCharsTask complete.result=" + content);  
  19. return content.toString();  
  20.     }  

这是一个正常的短时的任务,产生10个随机字符组成的字符串。

1.测试invokeAny()

第一种情况,向线程池提交2个耗时任务SleepSecondsCallable

[java] view plaincopy

  1. /**
  2. * 提交的任务集合,一旦有1个任务正常完成(没有抛出异常),会终止其他未完成的任务
  3. */
  4. public static void invokeAny1() throws Exception  
  5. {  
  6.     ExecutorService executorService = Executors.newFixedThreadPool(3);  
  7.     List<Callable<String>> tasks = new ArrayList<Callable<String>>();  
  8.     tasks.add(new SleepSecondsCallable("t1", 2));  
  9.     tasks.add(new SleepSecondsCallable("t2", 1));  
  10.     String result = executorService.invokeAny(tasks);  
  11.     System.out.println("result=" + result);  
  12.     executorService.shutdown();  

程序的执行结果是:返回t2线程的执行结果t2_SleepSecondsCallable_succes,同时t1抛出java.lang.InterruptedException: sleep interrupted。

也就说:一旦有1个任务正常完成(执行过程中没有抛异常),线程池会终止其他未完成的任务

第二种情况,向线程池提交3个异常任务ExceptionCallable

[java] view plaincopy

  1. /**
  2. * 没有1个正常完成的任务,invokeAny()方法抛出ExecutionException,封装了任务中元素的异常
  3. */
  4. public static void invokeAny2() throws Exception  
  5. {  
  6.     ExecutorService executorService = Executors.newFixedThreadPool(3);  
  7.     List<Callable<String>> tasks = new ArrayList<Callable<String>>();  
  8.     tasks.add(new ExceptionCallable());  
  9.     tasks.add(new ExceptionCallable());  
  10.     tasks.add(new ExceptionCallable());  
  11.     String result = executorService.invokeAny(tasks);  
  12.     System.out.println("result=" + result);  
  13.     executorService.shutdown();  

程序执行结果是:调用invokeAny()报错 java.util.concurrent.ExecutionException: java.lang.NullPointerException。

也就是说:如果提交的任务列表中,没有1个正常完成的任务,那么调用invokeAny会抛异常,究竟抛的是哪儿个任务的异常,无关紧要

第三种情况:先提交3个异常任务,再提交1个正常的耗时任务

[java] view plaincopy

  1. /**
  2. * 有异常的任务,有正常的任务,invokeAny()不会抛异常,返回最先正常完成的任务
  3. */
  4. public static void invokeAny3() throws Exception  
  5. {  
  6.     ExecutorService executorService = Executors.newFixedThreadPool(3);  
  7.     List<Callable<String>> tasks = new ArrayList<Callable<String>>();  
  8.     tasks.add(new ExceptionCallable());  
  9.     tasks.add(new ExceptionCallable());  
  10.     tasks.add(new ExceptionCallable());  
  11.     tasks.add(new ExceptionCallable());  
  12.     tasks.add(new SleepSecondsCallable("t1", 2));  
  13.     String result = executorService.invokeAny(tasks);  
  14.     System.out.println("result=" + result);  
  15.     executorService.shutdown();  

程序执行结果是:不会抛出任何异常,打印出t2任务的返回结果。也就是说:invokeAny()和任务的提交顺序无关,只是返回最早正常执行完成的任务

第四种情况,测试下使用限时版本的invokeAny(),主要功能与不限时版本的差别不大

[java] view plaincopy

  1. /**
  2. * 还没有到超时之前,所以的任务都已经异常完成,抛出ExecutionException<br>
  3. * 如果超时前满,还没有没有完成的任务,抛TimeoutException
  4. */
  5. public static void invokeAnyTimeout() throws Exception  
  6. {  
  7.     ExecutorService executorService = Executors.newFixedThreadPool(3);  
  8.     List<Callable<String>> tasks = new ArrayList<Callable<String>>();  
  9.     tasks.add(new ExceptionCallable());  
  10.     tasks.add(new ExceptionCallable());  
  11.     tasks.add(new ExceptionCallable());  
  12.     tasks.add(new ExceptionCallable());  
  13.     String result = executorService.invokeAny(tasks, 2, TimeUnit.SECONDS);  
  14.     System.out.println("result=" + result);  
  15.     executorService.shutdown();  

程序执行结果是:抛出ExecutionException。这个其实很合理,也很好理解。如果在超时之前,所有任务已经都是异常终止,那就没有必要在等下去了;如果超时之后,仍然有正在运行或等待运行的任务,那么会抛出TimeoutException。

最后我们来看下,JDK源码中ExecutorService.invokeAny的方法签名和注释

[java] view plaincopy

  1. /**
  2.      * Executes the given tasks, returning the result
  3.      * of one that has completed successfully (i.e., without throwing
  4.      * an exception), if any do. Upon normal or exceptional return,
  5.      * tasks that have not completed are cancelled.
  6.      * The results of this method are undefined if the given
  7.      * collection is modified while this operation is in progress.
  8.      *
  9.      * @param tasks the collection of tasks
  10.      * @return the result returned by one of the tasks
  11.      * @throws InterruptedException if interrupted while waiting
  12.      * @throws NullPointerException if tasks or any of its elements
  13.      *         are <tt>null</tt>
  14.      * @throws IllegalArgumentException if tasks is empty
  15.      * @throws ExecutionException if no task successfully completes
  16.      * @throws RejectedExecutionException if tasks cannot be scheduled
  17.      *         for execution
  18.      */
  19.     <T> T invokeAny(Collection<? extends Callable<T>> tasks)  
  20. throws InterruptedException, ExecutionException; 

与我们测试结果一致,invokeAny()返回最先正常完成(without throwing exception)的任务直接结果;一旦有任务正常完成或者调用出现异常,线程池都会终止正在运行或等待运行(tasks that have not completed are cancelled)的任务。

2.测试invokeAll()

这个方法相对来说比较好理解,就是执行任务列表中的所有任务,并返回与每个任务对应的Futue。也就是说,任务彼此之间不会相互影响,可以通过future跟踪每一个任务的执行情况,比如是否被取消,是正常完成,还是异常完成,这主要使用Future类提供的API。

[java] view plaincopy

  1. public static void testInvokeAll() throws Exception  
  2. {  
  3.     ExecutorService executorService = Executors.newFixedThreadPool(5);  
  4.     List<Callable<String>> tasks = new ArrayList<Callable<String>>();  
  5.     tasks.add(new SleepSecondsCallable("t1", 2));  
  6.     tasks.add(new SleepSecondsCallable("t2", 2));  
  7.     tasks.add(new RandomTenCharsTask());  
  8.     tasks.add(new ExceptionCallable());  
  9. // 调用该方法的线程会阻塞,直到tasks全部执行完成(正常完成/异常退出)
  10.     List<Future<String>> results = executorService.invokeAll(tasks);  
  11. // 任务列表中所有任务执行完毕,才能执行该语句
  12.     System.out.println("wait for the result." + results.size());  
  13.     executorService.shutdown();  
  14. for (Future<String> f : results)  
  15.     {  
  16. // isCanceled=false,isDone=true
  17.         System.out.println("isCanceled=" + f.isCancelled() + ",isDone="
  18.                 + f.isDone());  
  19. // ExceptionCallable任务会报ExecutionException
  20.         System.out.println("task result=" + f.get());  
  21.     }  

程序的执行结果和一些结论,已经直接写在代码注释里面了。invokeAll是一个阻塞方法,会等待任务列表中的所有任务都执行完成。不管任务是正常完成,还是异常终止,Future.isDone()始终返回true。通过Future.isCanceled()可以判断任务是否在执行的过程中被取消。通过Future.get()可以获取任务的返回结果,或者是任务在执行中抛出的异常。

第二种情况,测试限时版本的invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)

[java] view plaincopy

  1. /**
  2. * 可以通过Future.isCanceled()判断任务是被取消,还是完成(正常/异常)<br>
  3. * Future.isDone()总是返回true,对于invokeAll()的调用者来说,没有啥用
  4. */
  5. public static void testInvokeAllTimeout() throws Exception  
  6. {  
  7.     ExecutorService executorService = Executors.newFixedThreadPool(5);  
  8.     List<Callable<String>> tasks = new ArrayList<Callable<String>>();  
  9.     tasks.add(new SleepSecondsCallable("t1", 2));  
  10.     tasks.add(new SleepSecondsCallable("t2", 2));  
  11.     tasks.add(new SleepSecondsCallable("t3", 3));  
  12.     tasks.add(new RandomTenCharsTask());  
  13.     List<Future<String>> results = executorService.invokeAll(tasks, 1,  
  14.             TimeUnit.SECONDS);  
  15.     System.out.println("wait for the result." + results.size());  
  16. for (Future<String> f : results)  
  17.     {  
  18.         System.out.println("isCanceled=" + f.isCancelled() + ",isDone="
  19.                 + f.isDone());  
  20.     }  
  21.     executorService.shutdown();  

执行结果是:

wait for the result.4
isCanceled=true,isDone=true
isCanceled=true,isDone=true
isCanceled=true,isDone=true
isCanceled=false,isDone=true

也就是说给定的超时期满,还没有完成的任务会被取消,即Future.isCancelled()返回true;在超时期之前,无论是正常完成还是异常终止的任务,Future.isCancelled()返回false。
第三种情况,测试在等待invokeAll执行完成之前,线程被中断。

[java] view plaincopy

  1. /**
  2. * 如果线程在等待invokeAll()执行完成的时候,被中断,会抛出InterruptedException<br>
  3. * 此时线程池会终止没有完成的任务,这主要是为了减少资源的浪费.
  4. */
  5. public static void testInvokeAllWhenInterrupt() throws Exception  
  6. {  
  7. final ExecutorService executorService = Executors.newFixedThreadPool(5);  
  8. // 调用invokeAll的线程
  9.     Thread invokeAllThread = new Thread() {  
  10. @Override
  11. public void run()  
  12.         {  
  13.             List<Callable<String>> tasks = new ArrayList<Callable<String>>();  
  14.             tasks.add(new SleepSecondsCallable("t1", 2));  
  15.             tasks.add(new SleepSecondsCallable("t2", 2));  
  16.             tasks.add(new RandomTenCharsTask());  
  17. // 调用线程会阻塞,直到tasks全部执行完成(正常完成/异常退出)
  18. try
  19.             {  
  20.                 List<Future<String>> results = executorService  
  21.                         .invokeAll(tasks);  
  22.                 System.out.println("wait for the result." + results.size());  
  23.             } catch (InterruptedException e)  
  24.             {  
  25.                 System.out  
  26.                         .println("I was wait,but my thread was interrupted.");  
  27.                 e.printStackTrace();  
  28.             }  
  29.         }  
  30.     };  
  31.     invokeAllThread.start();  
  32.     Thread.sleep(200);  
  33.     invokeAllThread.interrupt();  
  34.     executorService.shutdown();  

invokeAllThread 线程调用了ExecutorService.invokeAll(),在等待任务执行完成的时候,invokeAllThread被别的线程中断了。这个时候,

ExecutorService.invokeAll()会抛出java.lang.InterruptedException,任务t1和t2都被终止抛出java.lang.InterruptedException: sleep interrupted。

也就是说一旦ExecutorService.invokeAll()方法产生了异常,线程池中还没有完成的任务会被取消执行。

多线程(二)