Прерывать все потоки, если исключение происходит в любом из потоков из списка нитей

Question

Я вызываю Список потоков с помощью invokeAll(). AFAIK invokeAll() вернется только тогда, когда все потоки завершат свою задачу.

ExecutorService threadExecutor = Executors.newFixedThreadPool(getThreadSize()); 
List<Future<Object>> future = w_threadExecutor.invokeAll(threadList); 

это называется, когда все нити заканчивают

for (Future<Object> w_inProgressThread : w_future) 
{ 
// 

Это останавливает поток, в котором исключение Произошло, а не оставшиеся один. Есть ли способ остановить все остальные потоки, если какое-либо из исключений потока выдает? Или мне нужно отправить каждую задачу вместо invokeAll() ??

Я попытался использовать invokeAny() вместо invokeAll(), но не отменил оставшуюся задачу. invokeAny(): Если одна из задач завершена (или выбрасывает исключение), остальная часть Callable отменяется. Ref: http://tutorials.jenkov.com/java-util-concurrent/executorservice.html

Update:

CompletionService<Object> completionService = new ExecutorCompletionService<Object>(w_threadExecutor); 
       List<Future<Object>> futures = new ArrayList<Future<Object>>(); 
       for(Thread w_mt : threadList) 
       { 
       futures.add(completionService.submit(w_mt)); 
       } 
       for (int numTaken = 0; numTaken < futures.size(); numTaken++) { 
        Future f = completionService.take(); 
        try { 
         Object result = f.get(); 
         System.out.println(result); // do something with the normal result 
        } catch (Exception e) { 
         System.out.println("Catched ExecutionException, shutdown now!"); 
         //threadExecutor.shutdownNow(); 
         Thread.currentThread().interrupt(); 

         for (Future<Object> inProgressThread : futures) 
         { 
          inProgressThread.cancel(true); 
         } 
         break; 
        } 

Update 1:

Как было предложено waltersu я попытался

ExecutorService threadExecutor = Executors.newFixedThreadPool(3); 
       CompletionService<Object> completionService = new ExecutorCompletionService<Object>(threadExecutor); 
       List<Future<Object>> futures = new ArrayList<Future<Object>>(); 
       futures.add(completionService.submit(new Callable<Object>() { 
       @Override 
       public Object call() throws Exception { 
        String s=null; 
       // Thread.sleep(1000); 
        for(int i=0; i < 1000000; i++){ 
         int j =10 ; 
         if(i==100) 
         { 

         s.toString(); 
         } 

         System.out.println("dazfczdsa :: " + i); 
        } 
        //throw new Exception("This is an expected Exception"); 
       return s; 
       } 
       })); 
       futures.add(completionService.submit(new Callable<Object>() { 
       @Override 
       public Object call() throws Exception { 
        for(int i=0; i < 1000000; i++){ 
         int j =0 ; 
         j= j+2; 
         System.out.println("dasa :: " + i); 
        } 
        Thread.sleep(3000); 

        return "My First Result"; 
       } 
       })); 

       while (futures.size() > 0) { 
       Future f = completionService.take(); 
       futures.remove(f); 
       try { 
        Object result = f.get(); 
        System.out.println(result); // do something with the normal result 
       } catch (ExecutionException e) { 
        System.out.println("Caught exception from one task: " + e.getCause().getMessage() + ". shutdown now!"); 
        f.cancel(true); 
        threadExecutor.shutdownNow(); 
        break; 
       } 
       } 
       System.out.println("Main exists"); 

это не останавливается, когда происходит исключение

Answer 1

for(Future<Object> fut : future){ 
    try { 
      System.out.println(fut.get()); 
    } 
    catch (InterruptedException | ExecutionException e) { 
     Thread.currentThread().interrupt(); 
     // or notify all here 
    } 

Answer 2

Вы можете использовать Wrappers, которые берут ваши рабочие runnables и ссылку на Executor, на которые они будут вызваны. В случае исключения вы можете вызвать немедленное завершение работы внутри run().

class ExceptionHandlingWrapper implements Runnable{ 
    private ExecutorService es; 
    private Runnable childRunnable; 

    // CTOR taking an ExecutorService and a Runnable 
    public ExceptionHandlingWrapper (ExecutorService es, Runnable work){ 
     this.es = es; 
     this.childRunnable = work; 
    } 

    @Override public void run(){ 
     try{ 
      childRunnable.run(); 
     } 
     catch(Exception ex){ 
      // Todo: LOG IT! 
      es.shutdownNow(); 
     } 
    } 
} 

Это также работает с Callables, конечно.

Answer 3

Вы должны представить() один на один, а не invokeAll(), а затем проверить, если Future имеет исключение.

public static void main(String[] args) throws InterruptedException { 
    ExecutorService threadExecutor = Executors.newFixedThreadPool(3); 
    CompletionService<Object> completionService = new ExecutorCompletionService<>(threadExecutor); 
    List<Future<Object>> futures = new ArrayList<>(); 
    futures.add(completionService.submit(new Callable<Object>() { 
    @Override 
    public Object call() throws Exception { 
     Thread.sleep(1000); 
     throw new Exception("This is an expected Exception"); 
    } 
    })); 
    futures.add(completionService.submit(new Callable<Object>() { 
    @Override 
    public Object call() throws Exception { 
     Thread.sleep(3000); 
     return "My First Result"; 
    } 
    })); 

    while (futures.size() > 0) { 
    Future f = completionService.take(); 
    futures.remove(f); 
    try { 
     Object result = f.get(); 
     System.out.println(result); // do something with the normal result 
    } catch (ExecutionException e) { 
     System.out.println("Caught exception from one task: " + e.getCause().getMessage() + ". shutdown now!"); 
     threadExecutor.shutdownNow(); 
     break; 
    } 
    } 
    System.out.println("Main exists"); 
} 

---

Update 1: (Чтобы ответить на обновление 1 вопрос Op в)

Это потому, что ваша задача имеет длинную петлю, которая не проверяет прерывания, что делает ваша задача noncancelable. Тогда как вы это прекратите? Я думаю, вам нужно изменить свои другие задачи, чтобы сделать их отменными. Как the official doc говорит:

Что делать, если поток длится много времени, не вызывая метод, который вызывает InterruptedException? Затем он должен периодически вызывать Thread.interrupted, который возвращает true, если получено прерывание. Например:

for (int i = 0; i < inputs.length; i++) { 
    heavyCrunch(inputs[i]); 
    if (Thread.interrupted()) { 
     // We've been interrupted: no more crunching. 
     return; 
    } 
} 

Что делать, если вы не хотите, чтобы изменить свою задачу, а также хотите, чтобы быстро остановить? Существует способ, которым вы можете остановить неразрешимый поток. Это Thread.stop().Но сначала вы не можете получить поток из threadpool без использования рефлексии. Кроме того, он устарел, потому что «он по своей сути небезопасен» в соответствии с javadoc.

Итак, лучшая практика (я думаю) заключается в проверке прерывания вашей задачи (или части кода), которая не подлежит проверке и длится долгое время.

Answer 4

Несколько предложений:

1. Удалить invokeAll и представить все задачи ExecutorService.
2. поймать исключение из ГЭТ() на будущее и использовать shutdown & shutdownNow, как объяснено в этих вопросах:

How to stop next thread from running in a ScheduledThreadPoolExecutor

How to forcefully shutdown java ExecutorService