How to retrieve and handle exceptions in Java's ExecutorService

Question

I am trying to figure out a way to handle exceptions in a multi-thread setting. I would like to execute certain tasks in parallel, each of which might throw an exception that I need to react to (basically, by putting the failed task back into an execution queue). However, it seems to only way to actually get the exception from the thread is to create a Future and call its get() method. However, this essentially turns the calls into synchronous calls.

Maybe some code will illustrate the point:

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 
} 

However, in this case all tasks are launched, but the exceptions don't seem to get caught in this catch block here.

An alternative would be to use a Future instead of a thread and retrieve its result:

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

In this case, the exceptions are caught alright in the catch block, but at the price of having to wait until this operation is finished. So, the tasks are executed sequentially and not in parallel, as desired.

What am I missing? How can catch each's tasks Exceptions and react to them?

Answer 1

you could trigger all your tasks within one loop and check/await/retry in another:

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