Catching Ausnahmen aller Futures/Callables in einem ExecutorService

Question

Ich habe eine ExecutorService enthält eine Reihe von Callables ausgeführt. Ich habe eine Liste von Futures für diese Callables. Ich möchte so schnell wie möglich herausfinden, ob einer der Callables einen Exception wirft. Alle Callables sind gleichermaßen wahrscheinlich zu werfen Exception. Die Callables laufen normalerweise für mehrere Stunden.

Der übliche Ansatz scheint die Methode Future.get() zu verwenden. Sie können dies jedoch nur für einen Future verwenden. Wenn ein anderer Future einen Exception wirft, werde ich nicht benachrichtigt. Also dachte ich darüber nach, eine Schleife zu schreiben, die die Methode Future.isDone() für alle Futures prüft und nach jeder Iteration für einige Zeit schläft. Allerdings ist dieser Ansatz nicht wirklich nett, also habe ich mich gefragt, ob es einen besseren Weg dafür gibt?

Answer 1

Sie sollten eine ExecutorCompletionService verwenden, wickeln Sie Ihren Executor damit und dann ruft #take() das erste abgeschlossene Ergebnis zurück.

Beispiel:

CompletionService<Object> completionService = new ExecutorCompletionService<>(executor); 
//submit a bunch of tasks 
IntStream.range(0, 100).forEach(i-> completionService.submit(Object::new)); 
//Wait for any of them to complete, wrap in a loop to take them all 
completionService.take(); 

Answer 2

Sie CompletableFuture für Ihren Anwendungsfall

static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) 

Gibt ein neues CompletableFuture, die abgeschlossen wird, wenn eine der gegeben CompletableFutures abgeschlossen ist, mit dem gleichen verwenden können Ergebnis. Andernfalls, wenn ausnahmsweise abgeschlossen wurde, führt die zurückgegebene CompletableFuture auch , so dass eine CompletionException diese Ausnahme als Ursache hat. Wenn keine CompletableFutures bereitgestellt werden, wird eine unvollständige CompleableFuture zurückgegeben.

Sie sollten alle Ihre Zukünfte in einer Liste speichern. `

Jetzt

List<CompletableFuture<?>> futureList = // all futures; 
while(futureList.size()>0){ 
    CompletableFuture<?> completed = CompletableFuture.anyOf(futureList); //convert futureList to array 
    if(completed.isCompletedExceptionally()){ 
     // first future that completed with an exception 
    }else{ 
     //future completed without exception, store the result or process 
     futureList.remove(completed); // remove it from the incomplete list 
    } 
} 

Sie erhalten kann ein CompletableFuture wie

final CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { 
     //...long running... 
     return "returnVal"; 
     }, 
    executor); //here executor is your executor pool 

Wenn Sie nicht wollen, explizit Testamentsvollstrecker Pool

final CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { 
      //...long running... 
      return "returnVal"; 
      }); 

jedoch in diesem Fall, es zu benutzen wird an die ForkJoinPool .commonPool()

gesendet werden