如何在Java的ExecutorService中检索和处理异常

Question

我试图找出一种方法来处理多线程设置中的异常.我想并行执行某些任务,每个任务都可能抛出一个我需要做出反应的异常(基本上,通过将失败的任务放回执行队列).但是,实际上从线程中获取异常的唯一方法是创建Future并调用其get()方法.但是,这实际上将调用转换为同步调用.

也许一些代码会说明这一点:

ExecutorService executor = Executors.newFixedThreadPool(nThreads);
Task task = taskQueue.poll(); // let's assume that task implements Runnable
try {
  executor.execute(task);
}
catch(Exception ex) {
  // record the failed task, so that it can be re-added to the queue 
} 

但是,在这种情况下,所有任务都会启动,但这里的异常似乎没有被捕获到此catch块中.

另一种方法是使用Future而不是线程并检索其结果:

try {
  Future<?> future = executor.submit(task);
  future.get();
}
...

在这种情况下,异常会在catch块中被捕获,但代价是必须等到此操作完成.因此,根据需要,任务按顺序执行而不是并行执行.

我错过了什么？如何捕捉每个人的任务例外并对他们做出反应？

Answer 1

您可以在一个循环中触发所有任务，并在另一个循环中检查/等待/重试：

Map<Future<?>, Task> futures = new HashMap<Future<?>, Task>()
while(!taskQueue.isEmpty()){
    Task task = taskQueue.poll();
    Future<?> future = executor.submit(task);
    futures.put(future, task);
}

for(Map.Entry<Future<?>, Task> entry : futures.entrySet()){

    try {
        entry.getKey().get();
    }
    catch(ExecutionException ex) {
        // record the failed task, so that it can be re-added to the queue 
        // you should add a retry counter because you want to prevent endless loops
        taskQueue.add(entry.getValue());
    }
    catch(InterrupredException ex){ 
        // thread interrupted, exit
        Thread.interrupt();
        return;
    }
}

HTH，马克