java - 现在关闭 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 方法使主线程在第一个 Future 实例上调用 f.get()对应于长任务，这使得它等待任务的持续时间，所以无论发生什么至少 5 秒。一旦完成，它将在已经结束的第二个 Future 上调用 f.get()，因此它会立即从 中获取 CustomException >ExecutionException 并调用 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

正如您在输出中看到的那样，任务一很快就被中断了。

---

The message "Received:Error on task 2" was not thrown, so it looks like a successful execution, and is not the case

不，这只是因为第一次调用 f.get() 会按预期抛出 InterruptedException，这使得它从 executeFutures 中退出，因为catch 在循环外执行，将它移到循环内，如下所示:

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