等待未来的名单

Question

我有一个返回List期货的方法

List<Future<O>> futures = getFutures();

现在我想要等到所有期货都成功处理完成,或者将来输出的任何任务抛出异常.即使一个任务抛出异常,也没有必要等待其他期货.

简单的方法是

wait() {

   For(Future f : futures) {
     try {
       f.get();
     } catch(Exception e) {
       //TODO catch specific exception
       // this future threw exception , means somone could not do its task
       return;
     }
   }
}

但问题是,例如,如果第四个未来抛出异常,那么我将不必要地等待前3个期货可用.

怎么解决这个？会以任何方式倒数闩锁帮助吗？我无法使用Future,isDone因为java doc说

boolean isDone()
Returns true if this task completed. Completion may be due to normal termination, an exception, or cancellation -- in all of these cases, this method will return true.

Answer 1

您可以使用CompletionService在它们准备好后立即接收期货,如果其中一个抛出异常,则取消处理.像这样的东西:

Executor executor = Executors.newFixedThreadPool(4);
CompletionService<SomeResult> completionService = 
       new ExecutorCompletionService<SomeResult>(executor);

//4 tasks
for(int i = 0; i < 4; i++) {
   completionService.submit(new Callable<SomeResult>() {
       public SomeResult call() {
           ...
           return result;
       }
   });
}

int received = 0;
boolean errors = false;

while(received < 4 && !errors) {
      Future<SomeResult> resultFuture = completionService.take(); //blocks if none available
      try {
         SomeResult result = resultFuture.get();
         received ++;
         ... // do something with the result
      }
      catch(Exception e) {
             //log
         errors = true;
      }
}

我认为如果其中一个任务抛出错误,您可以进一步改进以取消任何仍在执行的任务.

编辑:我在这里找到了一个更全面的例子:http://blog.teamlazerbeez.com/2009/04/29/java-completionservice/

Answer 2

如果您使用的是Java 8,则可以使用CompletableFuture和CompletableFuture.allOf更轻松地完成此操作,仅在完成所有提供的CompletableFutures之后才应用回调.

// Waits for *all* futures to complete and returns a list of results.
// If *any* future completes exceptionally then the resulting future will also complete exceptionally.

public static <T> CompletableFuture<List<T>> all(List<CompletableFuture<T>> futures) {
    CompletableFuture[] cfs = futures.toArray(new CompletableFuture[futures.size()]);

    return CompletableFuture.allOf(cfs)
            .thenApply(ignored -> futures.stream()
                                    .map(CompletableFuture::join)
                                    .collect(Collectors.toList())
            );
}

Answer 3

您可以使用ExecutorCompletionService.文档甚至还有一个确切用例的示例:

相反,假设您希望使用任务集的第一个非null结果,忽略任何遇到异常,并在第一个任务准备就绪时取消所有其他任务:

void solve(Executor e, Collection<Callable<Result>> solvers) throws InterruptedException {
    CompletionService<Result> ecs = new ExecutorCompletionService<Result>(e);
    int n = solvers.size();
    List<Future<Result>> futures = new ArrayList<Future<Result>>(n);
    Result result = null;
    try {
        for (Callable<Result> s : solvers)
            futures.add(ecs.submit(s));
        for (int i = 0; i < n; ++i) {
            try {
                Result r = ecs.take().get();
                if (r != null) {
                    result = r;
                    break;
                }
            } catch (ExecutionException ignore) {
            }
        }
    } finally {
        for (Future<Result> f : futures)
            f.cancel(true);
    }

    if (result != null)
        use(result);
}

需要注意的重要一点是ecs.take()将获得第一个完成的任务,而不仅仅是第一个提交的任务.因此,您应该按照完成执行(或抛出异常)的顺序获取它们.

Answer 4

CompletableFuture在Java 8中使用

    // Kick of multiple, asynchronous lookups
    CompletableFuture<User> page1 = gitHubLookupService.findUser("Test1");
    CompletableFuture<User> page2 = gitHubLookupService.findUser("Test2");
    CompletableFuture<User> page3 = gitHubLookupService.findUser("Test3");

    // Wait until they are all done
    CompletableFuture.allOf(page1,page2,page3).join();

    logger.info("--> " + page1.get());