立即關閉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