立即关闭ExecutionException

Question

我读了很多关于ExecutorService的帖子，但我无法找到我需要的方式。

我需要一些并发线程。当他们中的任何一个抛出自定义异常时，所有剩余的任务都被取消。

这是我做的一个例子。该任务正在并发工作，但不会在例外情况下中断。

public class Main { 

    public static void main(String[] args) { 
     ExecutorService executorService = Executors.newFixedThreadPool(2); 
     List<Future> futures = new ArrayList<Future>(); 

     futures.add(executorService.submit(new Callable<Void>() { 
      public Void call() throws Exception { 
       Thread.sleep(5000); 
       System.out.println("Task 1 done"); 
       return null; 
      } 
     })); 

     futures.add(executorService.submit(new Callable<Void>() { 
      public Void call() throws Exception { 
       Thread.sleep(2000); 
       System.out.println("Task 2 done"); 
       if (true) { 
        throw new CustomException("Error on task 2"); 
       } 
       return null; 
      } 
     })); 
     executorService.shutdown(); 

     try { 
      executeFutures(futures); 
     } catch (CustomException ex) { 
      System.out.println("Received:" + ex.getMessage()); 
      executorService.shutdownNow(); 
     }  
    } 

    private static void executeFutures(List<Future> futures) throws CustomException { 
     try { 
      for (Future f : futures) { 
       f.get(); 
      } 
     } catch (ExecutionException | InterruptedException e) { 
      if (e.getCause() instanceof CustomException) { 
       throw (CustomException) e.getCause(); 
      } 
     } 
    }  
} 

这是输出：

Task 2 done //exception is thrown here but task1 continue. 
Task 1 done 
Received:Error on task 2 

任何帮助将不胜感激。

Answer 1

你的问题是由于该方法executeFutures使主线程调用f.get()对应于长期的任务，这使得它等待任务的持续时间如此至少5秒钟，无论发生什么情况第一Future实例。一旦完成，它将在第二个Future上调用f.get()，这已经结束，因此它立即从ExecutionException得到CustomException，并调用executorService.shutdownNow()，但由于没有更多任务需要中断，所以已经太晚了。

你能做什么，是使用Callable类型的装饰器，当CustomException被抛出时会自动关闭线程池，这样线程池就会直接被执行引发任务的线程关闭异常而不是使用主线程。

事情是这样的：

public class AutoShutdown<V> implements Callable<V> { 

    private final ExecutorService executorService; 
    private final Callable<V> task; 

    public AutoShutdown(final ExecutorService executorService, final Callable<V> task) { 
     this.executorService = executorService; 
     this.task = task; 
    } 

    @Override 
    public V call() throws Exception { 
     try { 
      return task.call(); 
     } catch (CustomException e) { 
      executorService.shutdownNow(); 
      throw e; 
     } 
    } 
} 

然后，你将需要通过为旁边的装饰提交您的任务：

futures.add(
    executorService.submit(
     new AutoShutdown<>(
      executorService, 
      new Callable<Void>() { 
       public Void call() throws Exception { 
        Thread.sleep(5000); 
        System.out.println("Task 1 done"); 
        return null; 
       } 
      } 
     ) 
    ) 
); 

futures.add(
    executorService.submit(
     new AutoShutdown<>(
      executorService, 
      new Callable<Void>() { 
       public Void call() throws Exception { 
        Thread.sleep(2000); 
        System.out.println("Task 2 done"); 
        if (true) { 
         throw new CustomException("Error on task 2"); 
        } 
        return null; 
       } 
      } 
     ) 
    ) 
); 

输出：

Task 2 done 

，你可以在输出中看到，任务已经很快中断了。

---

消息“Received:Error on task 2”没有抛出，所以它看起来 像一个成功执行，而不是

不，它只是因为f.get()第一调用抛出的情况下InterruptedException如预期的那样使其从executeFutures退出，因为捕获是在循环外部执行的，因此将其在循环内部移动如下：

private static void executeFutures(List<Future> futures) throws CustomException { 
    for (Future f : futures) { 
     try { 
      f.get(); 
     } catch (ExecutionException | InterruptedException e) { 
      if (e.getCause() instanceof CustomException) { 
       throw (CustomException) e.getCause(); 
      } 
     } 
    } 
} 

输出：

Task 2 done 
Received:Error on task 2