java 如何使用 ExecutorService 和 Futures 处理 TimeoutException？

How to handle TimeoutException with ExecutorService and Futures?

提问by Brad

I have the following example code and let's say that MyCallable("B")takes longer than one second to execute, when the others execute quicker than one second. Therefore inside my loop that calls Future.get(), it will throw a TimeoutException.

我有以下示例代码，假设MyCallable("B")执行时间超过一秒，而其他代码的执行速度超过一秒。因此在我调用的循环中Future.get()，它会抛出一个TimeoutException.

public static void main(String[] args) {
    ExecutorService es = Executors.newFixedThreadPool(2);

    List<Future<String>> futures = new ArrayList<Future<String>>();

    futures.add(es.submit(new MyCallable("A")));
    futures.add(es.submit(new MyCallable("B")));
    futures.add(es.submit(new MyCallable("C")));
    futures.add(es.submit(new MyCallable("D")));
    futures.add(es.submit(new MyCallable("E")));

    try {
        for(Future<String> f  : futures) {
            try {
                System.out.println("result " + f.get(1, TimeUnit.SECONDS));
            }
            catch (TimeoutException e) {
                // how do I know which MyCallable() has timed out?
            } catch (ExecutionException e) {
                System.out.println(e.getMessage());
            }
        }
    }
    catch (InterruptedException e) {
        e.printStackTrace();
    }
    finally {
        es.shutdown();
    }
}

As expected each of the MyCallable() instances execute, but for the one that times out I would like to perform some error handling and this requires knowing which Callableis associated with which Future.

正如预期的那样，每个 MyCallable() 实例都执行，但对于超时的实例，我想执行一些错误处理，这需要知道哪个Callable与哪个Future.

Is there a mechanism for this association or is it up to my Callableto handle all the error processing inside it's call()method?

是否有这种关联的机制，还是由我Callable来处理它的call()方法中的所有错误处理？

采纳答案by Joe K

Seems like you could simply maintain a Map<Future<String>, Callable<String>>instead of a List<Future<String>>and retrieve the original Callable that way.

似乎您可以简单地维护 aMap<Future<String>, Callable<String>>而不是 aList<Future<String>>并以这种方式检索原始 Callable 。

If you want to get really clever, you could do it OO-style and extend ThreadPoolExecutor and create a Future decorator class. I think this is probably overkill, but you could do it like this:

如果你想变得非常聪明，你可以做 OO 风格并扩展 ThreadPoolExecutor 并创建一个 Future 装饰器类。我认为这可能有点矫枉过正，但你可以这样做：

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;


public class FutureWithCallable<T> implements Future<T> {
    private final Callable<T> callable;
    private final Future<T> wrapped;

    public FutureWithCallable(Future<T> wrapped, Callable<T> callable) {
        this.callable = callable;
        this.wrapped = wrapped;
    }

    public Callable<T> getCallable() {
        return callable;
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return wrapped.cancel(mayInterruptIfRunning);
    }

    @Override
    public T get() throws InterruptedException, ExecutionException {
        return wrapped.get();
    }

    @Override
    public T get(long timeout, TimeUnit unit) throws InterruptedException,
            ExecutionException, TimeoutException {
        return wrapped.get(timeout, unit);
    }

    @Override
    public boolean isCancelled() {
        return wrapped.isCancelled();
    }

    @Override
    public boolean isDone() {
        return wrapped.isDone();
    }
}

And then:

接着：

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

    public class ExecutorServiceWithCallable extends ThreadPoolExecutor {

        public ExecutorServiceWithCallable(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
            super(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue);
        }

        @Override
        public <T> FutureWithCallable submit(Callable<T> callable) {
            Future<T> future = super.submit(callable);
            return new FutureWithCallable<T>(future, callable);
        }

    }

回答by pankaj

public class TimeoutException extends Exception Exception thrown when a blocking operation times out. Blocking operations for which a timeout is specified need a means to indicate that the timeout has occurred. For many such operations it is possible to return a value that indicates timeout; when that is not possible or desirable then TimeoutException should be declared and thrown.

公共类 TimeoutException 扩展了阻塞操作超时时抛出的异常异常。指定超时的阻塞操作需要一种方法来指示超时已经发生。对于许多这样的操作，可以返回一个表示超时的值；如果这是不可能的或不可取的，则应声明并抛出 TimeoutException。