Come recuperare e gestire le eccezioni in ExecutorService di Java

Question

Sto cercando di trovare un modo per gestire le eccezioni in un'impostazione multi-thread. Vorrei eseguire alcune attività in parallelo, ognuna delle quali potrebbe generare un'eccezione alla quale ho bisogno di reagire (in pratica, inserendo l'attività fallita in una coda di esecuzione). Tuttavia, sembra che solo il modo per ottenere effettivamente l'eccezione dal thread è creare un futuro e chiamare il suo metodo get(). Tuttavia, questo essenzialmente trasforma le chiamate in chiamate sincrone.

Forse qualche codice illustrerà il punto:

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 
} 

Tuttavia, in questo caso, tutte le attività vengono lanciati, ma le eccezioni non sembrano farsi prendere in questo blocco catch qui.

Un'alternativa sarebbe quella di utilizzare un futuro al posto di un filo e recuperare il suo risultato:

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

In questo caso, le eccezioni sono catturati bene nel blocco catch, ma al prezzo di dover aspettare fino a quando questa operazione è finita. Quindi, i compiti vengono eseguiti in sequenza e non in parallelo, come desiderato.

Cosa mi manca? Come si possono cogliere le attività di ciascuna eccezione e reagire a loro?

Answer 1

si potrebbe innescare tutte le attività all'interno di un loop e controllare/await/riprovare in un altro:

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