在线程失败和异常处理时停止 ExecutorService

Question

这是我为暴露我的问题所做的一个简化示例。我有一些任务doSomeWork()，我使用 ExecutorService 以多线程方式处理（一次最多 4 个线程）。但是，如果任何线程/任务产生异常，我想：

1. 停止处理任何进一步的任务。

2. 在主线程级别捕获异常。

   public static void main(String[] args) {
       final ExecutorService threadPool = Executors.newFixedThreadPool(4);
       final ExecutorCompletionService<Void> completionService = new ExecutorCompletionService<>(threadPool);
   
       try {
           for (int i = 0; i < 10; i++) {
               int b = i;
                   completionService.submit(() -> doSomeWork(b));
           }
   
           threadPool.shutdown();
           threadPool.awaitTermination(8, TimeUnit.HOURS);
   
           System.exit(0);
   
       } catch (Exception e) {
           System.out.println("Something wrong happened: " + e.getMessage());
       }
   
       System.exit(1);
   
   }
   
   //This function have 50% odds of throwing an exception
   public static Void doSomeWork(int i) throws Exception {
   
       Thread.sleep(500);
       if ((Math.random() > 0.5))
       {
           System.out.println("I have reached indice: " + i);
       }
       else
       {
           throw new Exception("I couldn't handle indice " + i);
       }
       return null;
   }
   

目前，执行会输出如下内容：

I have reached indice: 0
I have reached indice: 2
I have reached indice: 1
I have reached indice: 4
I have reached indice: 6
I have reached indice: 7
I have reached indice: 5
I have reached indice: 9

如您所见，indice 3 丢失了，但剩余线程的执行已完成。它也没有输出任何关于异常的信息。

我想要的输出是这样的：

I have reached indice: 0
I have reached indice: 2
I have reached indice: 1
Something wrong happened: I couldn't handle indice 3

我围绕这个问题找到的其他解决方案是使用带有未来但以阻塞方式的可调用对象。我不能在等待未来的同时阻止其他线程的执行，否则整个多线程是没有意义的。

Answer 1

您可以使用CompletableFuture 做到这一点。这是我测试的主要功能：

final ExecutorService executorService = Executors.newFixedThreadPool(4);
final List<CompletableFuture<Void>> all = new ArrayList<>();

try {
    for (int i = 0; i < 10; i++) {
        int b = i;
        CompletableFuture<Void> v = CompletableFuture.runAsync(() -> {
                    try {
                        doSomeWork(b);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                },
                executorService);
        all.add(v);
    }

    CompletableFuture<Void> placeholder = CompletableFuture.allOf(all.toArray(new CompletableFuture[0]));
    failFast(all, placeholder);

    System.out.println("All tasks ended");

} catch (Exception e) {
    System.out.println("Something wrong happened: " + e.getMessage());
} finally {
    executorService.shutdownNow();
}

一旦其中一个失败（或全部完成），使联合未来失败的实用功能：

private static <T> void failFast(List<CompletableFuture<T>> futures, CompletableFuture<T> joint) {
    while (true) {
        if (joint.isDone()) {
            return;
        }
        for (CompletableFuture<T> future : futures) {
            if (future.isCompletedExceptionally()) {
                return;
            }
        }
    }
}

这是我得到的输出：

I have reached indice: 1
I have reached indice: 7
I have reached indice: 5
I have reached indice: 4
Something wrong happened: java.lang.RuntimeException: java.lang.Exception: I couldn't handle indice 0

解释：

CompletableFuture.runAsync() 方法允许您提供Runnable（您的doSomeWork）和具有一定数量线程的执行程序。在这里，我传递了一个有 4 个线程的执行程序（就像您在示例中所做的那样）。

在 runnable 内部，我不仅运行 doSomeWork 函数，而且还捕获 Exception 并抛出 RuntimeException（需要这样做，因为 Lambda 不支持检查异常，所以我需要包装它进入运行时，但它仍会中断执行并被您的主程序捕获）。

每次我为具有给定索引i 的任务创建新的CompletableFuture<Void> 时，我都会将此结果存储到可完成期货的列表中。

for 循环不会执行任何操作，因为可完成的期货是异步运行的。

因此，我与CompletableFuture.allOf(...) 创建了一个联合可完成的未来，然后我在这个未来上使用实用函数failFast，以便在其中一项任务失败时立即停止（或继续直到所有任务都完成）完成）。

因此，基本上只要其中一个期货未能引发异常，联合期货就被认为已完成，因此会将句柄留给您的主线程，与此同时，被抛出的 RuntimeException 击中在 lambda 表达式中。

注意：感谢 Thomas 的评论，我已更新代码以使用 ExecutorService 而不是简单的 Executor。这使您可以在捕获异常后在 finally 块内调用 shutdownNow()。 同样，正如 Thomas 所建议的，您可以直接在 doSomeWork 函数中抛出 RuntimeException，这样您就不需要在 lambda 表达式中捕获和包装。

其他说明： @matt 让我注意到了一些我不知道的事情。 .allOf() 未来将在 ALL 未来完成时完成，无论成功与否。 因此，正如他所指出的，我的解决方案不会按原样工作。我再次编辑了答案以考虑他的评论，感谢@matt 让我注意到。

Answer 2

听起来您基于不正确的假设排除了执行此操作的正确方法。保持你的未来。

List<Future<?>> futures = new ArrayList<>();

然后当你提交时。

futures.add( completionService.submit( () -> doSomeWork(b) ) );

现在，您可以在主线程中检查期货状态。

for(Future<?> f: futures){
    try{
        f.get();
    } catch( ExecutionException e){
        //execution exception handled on the main thread.
        completionService.shutdownNow();
    } catch( InterruptedException ie){
        //what should happen here.
    }
}

这样，shutdownNow 会被调用，因此所有未启动的任务都会返回并且不会启动。

您可以使用get 的超时来检查每个任务，因为有些任务将并行运行。

这是一个完整的可编译示例。

import java.util.concurrent.*;
import java.util.*;

public class ExecutorJunk{
    static int count = 0;
    static void task(){
        int z = count++;
        
        if(z == 3){
            throw new RuntimeException("z is 3");
        }
        
        System.out.println("z: " + z);
        try{ Thread.sleep(1500);} catch(InterruptedException e){};
    }
    
    public static void main(String[] args){
        ExecutorService service = Executors.newFixedThreadPool(4);
        
        List<Future<?>> all = new ArrayList<>();
        
        for(int i = 0; i<10; i++){
            all.add( service.submit(ExecutorJunk::task) );
        }
        service.shutdown();
        
        try{
            while(!service.isTerminated()){
                for(Future f: all){
                    try{
                      f.get(1, TimeUnit.MILLISECONDS);
                    } catch( TimeoutException toe){
                      //pass.
                    }
                }
            }
        } catch(Exception e){
            System.out.println( service.shutdownNow().size() + " tasks not started");
            e.printStackTrace();
        } 
    }
}

当我运行它时，我得到了。

z: 0
z: 1
z: 2
z: 4
5 个任务未开始
java.util.concurrent.ExecutionException: java.lang.RuntimeException: z 是 3
...

它可能会做得更聪明一点。比如在获取成功时清除期货列表，而不是使用超时，只需检查期货是否已完成，然后执行 future.get。