Wie Abfangen und Behandeln von Ausnahmen in Java ExecutorService

Question

Ich versuche, eine Möglichkeit zum Behandeln von Ausnahmen in einer Multithread-Einstellung. Ich möchte bestimmte Aufgaben parallel ausführen, von denen jede eine Ausnahme auslösen kann, auf die ich reagieren muss (im Grunde, indem ich die fehlgeschlagene Aufgabe in eine Ausführungswarteschlange zurückversetzt). Allerdings scheint es nur möglich zu sein, die Ausnahme vom Thread zu bekommen, indem man eine Future erstellt und ihre get() -Methode aufruft. Dies macht jedoch die Anrufe im Wesentlichen zu synchronen Anrufen.

Vielleicht werden einige Code, um den Punkt illustrieren:

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 
} 

jedoch in diesem Fall werden alle Aufgaben gestartet werden, aber die Ausnahmen scheinen nicht hier in diesem catch-Block einklemmen.

Eine Alternative eine Zukunft statt einem Thread zu verwenden wäre und sein Ergebnis abrufen:

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

In diesem Fall werden die Ausnahmen in Ordnung im catch-Block gefangen, aber um den Preis warten zu müssen bis dieser Vorgang beendet ist. Die Aufgaben werden also nacheinander und nicht wie gewünscht parallel ausgeführt.

Was fehlt mir? Wie kann man die Aufgaben jedes einzelnen Exceptions fangen und darauf reagieren?

Answer 1

Sie alle Ihre Aufgaben innerhalb einer Schleife auslösen könnte und prüfen/await/in ein anderes versuchen:

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, Mark