java并发编程框架 Executor ExecutorService invokeall

2022-10-24 19:58:02

首先介绍两个重要的接口，Executor和ExecutorService，定义如下：

  1. public interface Executor {
  2. void execute(Runnable command);
  3. }
  1. public interface ExecutorService extends Executor {
  2. //不再接受新任务，待所有任务执行完毕后关闭ExecutorService
  3. void shutdown();
  4. //不再接受新任务，直接关闭ExecutorService，返回没有执行的任务列表
  5. List<Runnable> shutdownNow();
  6. //判断ExecutorService是否关闭
  7. boolean isShutdown();
  8. //判断ExecutorService是否终止
  9. boolean isTerminated();
  10. //等待ExecutorService到达终止状态
  11. boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
  12. <T> Future<T> submit(Callable<T> task);
  13. //当task执行成功的时候future.get()返回result
  14. <T> Future<T> submit(Runnable task, T result);
  15. //当task执行成功的时候future.get()返回null
  16. Future<?> submit(Runnable task);
  17. //批量提交任务并获得他们的future，Task列表与Future列表一一对应
  18. <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
  19. throws InterruptedException;
  20. //批量提交任务并获得他们的future，并限定处理所有任务的时间
  21. <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
  22. long timeout, TimeUnit unit) throws InterruptedException;
  23. //批量提交任务并获得一个已经成功执行的任务的结果
  24. <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;
  25. <T> T invokeAny(Collection<? extends Callable<T>> tasks,
  26. long timeout, TimeUnit unit)
  27. throws InterruptedException, ExecutionException, TimeoutException;
  28. }

为了配合使用上面的并发编程接口，有一个Executors工厂类，负责创建各类满足ExecutorService接口的线程池，具体如下： 
newFixedThreadPool：创建一个固定长度的线程池，线程池中线程的数量从1增加到最大值后保持不变。如果某个线程坏死掉，将会补充一个新的线程。 
newCachedThreadPool：创建长度不固定的线程池，线程池的规模不受限制，不常用。 
newSingleThreadExecutor：创建一个单线程的Executor，他其中有一个线程来处理任务，如果这个线程坏死掉，将补充一个新线程。 
newScheduledThreadPool：创建固定长度的线程池，以延时或定时的方式来执行任务。 

下面是Executor和ExecutorService中常用方法的示例：

  1. import java.util.ArrayList;
  2. import java.util.Collection;
  3. import java.util.Iterator;
  4. import java.util.List;
  5. import java.util.concurrent.Callable;
  6. import java.util.concurrent.Executor;
  7. import java.util.concurrent.ExecutorService;
  8. import java.util.concurrent.Executors;
  9. import java.util.concurrent.Future;
  10. import java.util.concurrent.TimeUnit;
  11. public class Demo{
  12. public static void main(String [] args){
  13. //--------Executor示例------------//
  14. Executor s=Executors.newSingleThreadExecutor();
  15. s.execute(new MyRunnableTask("1"));
  16. //--------ExecutorService示例------------//
  17. ExecutorService es=Executors.newFixedThreadPool(2);
  18. //--------get()示例------------//
  19. Future<String> future=es.submit(new MyCallableTask("10"));
  20. try{
  21. System.out.println(future.get());
  22. }catch(Exception e){}
  23. //--------get(timeout, timeunit)示例------------//
  24. future=es.submit(new MyCallableTask("11"));
  25. try{
  26. System.out.println(future.get(500,TimeUnit.MILLISECONDS));
  27. }catch(Exception e){
  28. System.out.println("cancle because timeout");
  29. }
  30. //--------invokeAll(tasks)示例------------//
  31. List<MyCallableTask> myCallableTasks=new ArrayList<MyCallableTask>();
  32. for(int i=0;i<6;i++){
  33. myCallableTasks.add(new MyCallableTask(i+""));
  34. }
  35. try {
  36. List<Future<String>> results = es.invokeAll(myCallableTasks);
  37. Iterator<Future<String>> iterator=results.iterator();
  38. while(iterator.hasNext()){
  39. future=iterator.next();
  40. System.out.println(future.get());
  41. }
  42. } catch (Exception e) {}
  43. //--------invokeAll(tasks,timeout,timeunit))示例------------//
  44. try {
  45. //限定执行时间为2100ms,每个任务需要1000ms,线程池的长度为2,因此最多只能处理4个任务。一共6个任务，有2个任务会被取消。
  46. List<Future<String>> results = es.invokeAll(myCallableTasks,2100,TimeUnit.MILLISECONDS);
  47. Iterator<Future<String>> iterator=results.iterator();
  48. while(iterator.hasNext()){
  49. future=iterator.next();
  50. if(!future.isCancelled())
  51. System.out.println(future.get());
  52. else
  53. System.out.println("cancle because timeout");
  54. }
  55. } catch (Exception e) {}
  56. es.shutdown();
  57. }
  58. }
  59. class MyRunnableTask implements Runnable{
  60. private String name;
  61. public MyRunnableTask(String name) {
  62. this.name=name;
  63. }
  64. @Override
  65. public void run() {
  66. try {
  67. Thread.sleep(1000);
  68. } catch (InterruptedException e) {
  69. e.printStackTrace();
  70. }
  71. System.out.println("runnable task--"+name);
  72. }
  73. }
  74. class MyCallableTask implements Callable<String>{
  75. private String name;
  76. public MyCallableTask(String name) {
  77. this.name=name;
  78. }
  79. @Override
  80. public String call() throws Exception {
  81. try {
  82. Thread.sleep(1000);
  83. } catch (InterruptedException e) {}
  84. StringBuilder sb=new StringBuilder("callable task--");
  85. return sb.append(name).toString();
  86. }
  87. }

上面的ExecutorSerivce接口中的invokeAll(tasks)方法用于批量执行任务，并且将结果按照task列表中的顺序返回。此外，还存在一个批量执行任务的接口CompletionTask。ExecutorCompletionService是实现CompletionService接口的一个类，该类的实现原理很简单： 

用Executor类来执行任务，同时把在执行任务的Future放到BlockingQueue<Future<V>>队列中。该类实现的关键就是重写FutureTask类的done()方法，FutureTask类的done()方法是一个钩子函数（关于钩子函数，请读者自行查询），done()方法在FutureTask任务被执行的时候被调用。 

ExecutorCompletionService类的核心代码如下：

  1. public Future<V> submit(Runnable task, V result) {
  2. if (task == null) throw new NullPointerException();
  3. RunnableFuture<V> f = newTaskFor(task, result);
  4. executor.execute(new QueueingFuture(f));
  5. return f;
  6. }
  7. private class QueueingFuture extends FutureTask<Void> {
  8. QueueingFuture(RunnableFuture<V> task) {
  9. super(task, null);
  10. this.task = task;
  11. }
  12. protected void done() { completionQueue.add(task); }
  13. private final Future<V> task;
  14. }

其中的done()方法定义如下：

  1. /**
  2. * Protected method invoked when this task transitions to state
  3. * <tt>isDone</tt> (whether normally or via cancellation). The
  4. * default implementation does nothing.  Subclasses may override
  5. * this method to invoke completion callbacks or perform
  6. * bookkeeping. Note that you can query status inside the
  7. * implementation of this method to determine whether this task
  8. * has been cancelled.
  9. */
  10. protected void done() { }

ExecutorCompletionService的使用示例如下：

    1. import java.util.concurrent.Callable;
    2. import java.util.concurrent.CompletionService;
    3. import java.util.concurrent.ExecutionException;
    4. import java.util.concurrent.ExecutorCompletionService;
    5. import java.util.concurrent.Executors;
    6. import java.util.concurrent.Future;
    7. public class Demo{
    8. public static void main(String [] args) throws InterruptedException, ExecutionException{
    9. CompletionService<String> cs=new ExecutorCompletionService<String>(
    10. Executors.newFixedThreadPool(2));
    11. for(int i=0;i<6;i++){
    12. cs.submit(new MyCallableTask(i+""));
    13. }
    14. for(int i=0;i<6;i++){
    15. Future<String> future=cs.take();
    16. //Retrieves and removes the Future representing the next completed task,
    17. //waiting if none are yet present.
    18. System.out.println(future.get());
    19. }
    20. }
    21. }
    22. class MyCallableTask implements Callable<String>{
    23. private String name;
    24. public MyCallableTask(String name) {
    25. this.name=name;
    26. }
    27. @Override
    28. public String call() throws Exception {
    29. try {
    30. Thread.sleep(1000);
    31. } catch (InterruptedException e) {}
    32. StringBuilder sb=new StringBuilder("callable task--");
    33. return sb.append(name).toString();
    34. }
    35. }
上一篇：线上服务器TCP被打满是啥情况


下一篇：Linux查看机器是多少位