CompletableFuture：等待第一个正常返回？

问题描述

我有一些 CompletableFuture s我希望并行运行它们，等待第一个返回正常。

I have some CompletableFutures and I want to run them in parallel, waiting for the first that returns normally.

我知道我可以使用 CompletableFuture.anyOf 等待第一次返回，但这将返回通常或格外。我想忽略例外。

I know I can use CompletableFuture.anyOf to wait for the first to return, but this will return normally or exceptionally. I want to ignore exceptions.

List<CompletableFuture<?>> futures = names.stream().map(
  (String name) ->
    CompletableFuture.supplyAsync(
      () ->
        // this calling may throw exceptions.
        new Task(name).run()
    )
).collect(Collectors.toList());
//FIXME Can not ignore exceptionally returned takes.
Future any = CompletableFuture.anyOf(futures.toArray(new CompletableFuture<?>[]{}));
try {
    logger.info(any.get().toString());
} catch (Exception e) {
    e.printStackTrace();
}

推荐答案

您可以使用以下助手方法：

You may use the following helper method:

public static <T>
    CompletableFuture<T> anyOf(List<? extends CompletionStage<? extends T>> l) {

    CompletableFuture<T> f=new CompletableFuture<>();
    Consumer<T> complete=f::complete;
    l.forEach(s -> s.thenAccept(complete));
    return f;
}

您可以这样使用，以证明它会忽略先前的例外但是返回第一个提供的值：

which you can use like this, to demonstrate that it will ignore earlier exceptions but return the first provided value:

List<CompletableFuture<String>> futures = Arrays.asList(
    CompletableFuture.supplyAsync(
        () -> { throw new RuntimeException("failing immediately"); }
    ),
    CompletableFuture.supplyAsync(
        () -> { LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(5));
          return "with 5s delay";
        }),
    CompletableFuture.supplyAsync(
        () -> { LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(10));
          return "with 10s delay";
        })
);
CompletableFuture<String> c = anyOf(futures);
logger.info(c.join());

此解决方案的一个缺点是，如果永远不会完成>所有期货异常完成。一个解决方案，如果成功计算将提供第一个值，但如果根本没有成功计算则会异常失败，则需要更多参与：

One disadvantage of this solution is that it will never complete if all futures complete exceptionally. A solution, that will provide the first value if there is a successful computation but fail exceptionally if there is no successful computation at all, is a bit more involved:

public static <T>
    CompletableFuture<T> anyOf(List<? extends CompletionStage<? extends T>> l) {

    CompletableFuture<T> f=new CompletableFuture<>();
    Consumer<T> complete=f::complete;
    CompletableFuture.allOf(
        l.stream().map(s -> s.thenAccept(complete)).toArray(CompletableFuture<?>[]::new)
    ).exceptionally(ex -> { f.completeExceptionally(ex); return null; });
    return f;
}

它利用了 allOf 的特殊处理程序仅在所有期货完成后（异常或非特定）调用，并且未来只能完成一次（让这样的特殊内容... 除此之外）。当执行异常处理程序时，如果有结果，则完成任何使用结果完成未来的尝试都已完成，因此如果之前没有成功完成，则异常完成它的尝试只会成功。

It utilizes the fact that allOf’s exceptionally handler is only invoked after all futures have completed (exceptionally or not) and that a future can be completed only once (letting special things like obtrude… aside). When the exceptionally handler is executed, any attempt to complete the future with a result has been done, if there was one, so the attempt to complete it exceptionally only succeeds, if there was no previous successful completion.

它可以与第一个解决方案完全相同的方式使用，并且只有在所有计算都失败时才会表现出不同的行为，例如：

It can be used exactly the same way as the first solution and only exhibit different behavior if all computations fail, e.g.:

List<CompletableFuture<String>> futures = Arrays.asList(
    CompletableFuture.supplyAsync(
        () -> { throw new RuntimeException("failing immediately"); }
    ),
    CompletableFuture.supplyAsync(
        // delayed to demonstrate that the solution will wait for all completions
        // to ensure it doesn't miss a possible successful computation
        () -> { LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(5));
            throw new RuntimeException("failing later"); }
    )
);
CompletableFuture<String> c = anyOf(futures);
try { logger.info(c.join()); }
catch(CompletionException ex) { logger.severe(ex.toString()); }

上面的示例使用延迟证明解决方案将在没有成功时等待所有完成而将证明以后的成功将如何将结果变为成功。请注意，由于Ideones缓存结果，您可能不会注意到延迟。

The example above uses a delay demonstrating that the solution will wait for all completions when there is no success, whereas this example on ideone will demonstrate how a later success will turn the result into success. Note that due to Ideones caching of results you might not notice the delay.

请注意，如果所有期货都失败，则无法保证哪些例外情况会发生。得到报道。由于它在错误的情况下等待所有完成，任何可以使它达到最终结果。

Note that in the case that all futures fail, there is no guaranty about which of the exceptions will get reported. Since it waits for all completions in the erroneous case, any could make it to the final result.

这篇关于CompletableFuture：等待第一个正常返回？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！