Как остановить все запущенные потоки, если один из них выбрасывает исключение?

Ответ 1

Используйте ExecutorCompletionService, в комплекте с ExecutorService, который выделяет продолжительность задач (т.е. он не будет закрыт после этого):

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

class Threader {

    static ExecutorService service = Executors.newCachedThreadPool();

    public static void main(String[] args) {
        new Threader().start();
        service.shutdown();
    }

    private void start() {
        CompletionService<Void> completionService = new ExecutorCompletionService<Void>(
                service);
        /*
         * Holds all the futures for the submitted tasks
         */
        List<Future<Void>> results = new ArrayList<Future<Void>>();

        for (int i = 0; i < 3; i++) {
            final int callableNumber = i;

            results.add(completionService.submit(new Callable<Void>() {

                                                     @Override
                                                     public Void call() throws Exception {
                                                         System.out.println("Task " + callableNumber
                                                                 + " in progress");
                                                         try {
                                                             Thread.sleep(callableNumber * 1000);
                                                         } catch (InterruptedException ex) {
                                                             System.out.println("Task " + callableNumber
                                                                     + " cancelled");
                                                             return null;
                                                         }
                                                         if (callableNumber == 1) {
                                                             throw new Exception("Wrong answer for task "
                                                                     + callableNumber);
                                                         }
                                                         System.out.println("Task " + callableNumber + " complete");
                                                         return null;
                                                     }
                                                 }

            ));
        }

        boolean complete = false;
        while (!complete) {
            complete = true;
            Iterator<Future<Void>> futuresIt = results.iterator();
            while (futuresIt.hasNext()) {
                if (futuresIt.next().isDone()) {
                    futuresIt.remove();
                } else {
                    complete = false;
                }
            }

            if (!results.isEmpty()) {
                try {
                /*
                 * Examine results of next completed task
                 */
                    completionService.take().get();
                } catch (InterruptedException e) {
                /*
                 * Give up - interrupted.
                 */
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(e);
                } catch (ExecutionException e) {
                /*
                 * The task threw an exception
                 */
                    System.out.println("Execution exception " + e.getMessage());
                    complete = true;
                    for (Future<Void> future : results) {
                        if (!future.isDone()) {
                            System.out.println("Cancelling " + future);
                            future.cancel(true);
                        }
                    }
                }
            }
        }

    }
}

Вывод выглядит примерно так:

Task 0 in progress
Task 2 in progress
Task 1 in progress
Task 0 complete
Execution exception java.lang.Exception: Wrong answer for task 1
Cancelling [email protected]
Task 2 cancelled

где Задача 2 была отменена из-за отказа Задачи 1.

Ответ 2

Я настоятельно рекомендую вам использовать надежный механизм для подсчета защелки. Используйте всеохватывающие try-finally { latch.countDown(); } Обнаружение ошибок в потоках с использованием отдельного механизма.

Ответ 3

Думаю, вам нужно перестроить свой код. Взгляните на ExecutorService # invokeAny

Выполняет заданные задачи, возвращая результат завершено успешно (то есть, не выбрасывая исключение), если оно делать. При нормальном или исключительном возврате задачи, которые не завершены отменены. Результаты этого метода undefined, если заданный коллекция изменяется во время выполнения этой операции.

Это похоже на то, что вам нужно. И вам не нужен CountDownLatch, поскольку main будет блокироваться в invokeAny

Ответ 4

Я думаю, что вам понадобится еще один поток, назовите его "Watcher", который проверит значение AtomicBoolean для true. После его установки вы остановите основную службу выполнения. Имейте в виду, что механизм останова не гарантирует немедленную остановку всех потоков. Прочтите это, например: Изящное завершение потоков и исполнителей