在 Java 8 中并行调用两个函数

Question

我的 Spring Boot 应用程序中有一个用例如下：

我想使用以下函数从响应中获取 id 字段值：

String id = getIdFromResponse(response);

如果我在响应中没有得到任何 id，那么我使用以下函数检查请求参数中是否存在 id 字段：

String id = getIdFromRequest(request);

到目前为止，我正在按顺序调用它们。但我想让这两个函数并行运行，我想从它们中的任何一个获得 id 后立即停止。

我想知道是否有任何方法可以在 Java 8 中使用 streams 来实现这一点。

Answer 1

你可以这样使用：

String id = Stream.<Supplier<String>>of(
        () -> getIdFromResponse(response), 
        () -> getIdFromRequest(request)
    )
    .parallel()
    .map(Supplier::get)
    .filter(Objects::nonNull)
    .findFirst()
    .orElseThrow():

需要供应商，因为当你不使用它们时，两个请求仍然按顺序执行。

我还假设您的方法在找不到任何内容时返回 null，因此我必须使用 .filter(Objects::nonNull) 过滤掉这些值。

根据您的用例，您可以将 .orElseThrow() 替换为不同的内容，例如 .orElse(null)

Answer 2

只要有专门的方法，就不需要使用 Stream API。

ExecutorService::invokeAny(Collection<? extends Callable<T>>)

执行给定任务，返回已成功完成的任务的结果（即不抛出异常），如果有的话。正常或异常返回时，取消未完成的任务。

List<Callable<String>> collection = Arrays.asList(
    () -> getIdFromResponse(response),
    () -> getIdFromRequest(request)
);

// you want the same number of threads as the size of the collection
ExecutorService executorService = Executors.newFixedThreadPool(collection.size());
String id = executorService.invokeAny(collection);

三个注意事项：

• 如果没有及时得到结果，还有一个超时抛出的重载方法TimeoutException：invokeAny(Collection<? extends Callable<T>>, long, TimeUnit)
• 您需要通过invokeAny 方法处理ExecutionException 和InterruptedException。
• 完成后别忘了关闭服务

Answer 3

如果您想完全控制何时启用替代评估，您可以使用CompletableFuture：

CompletableFuture<String> job
    = CompletableFuture.supplyAsync(() -> getIdFromResponse(response));
String id;
try {
    id = job.get(300, TimeUnit.MILLISECONDS);
}
catch(TimeoutException ex) {
    // did not respond within the specified time, set up alternative
    id = job.applyToEither(
        CompletableFuture.supplyAsync(() -> getIdFromRequest(request)), s -> s).join();
}
catch(InterruptedException|ExecutionException ex) {
    // handle error
}

仅当第一个作业在指定时间内未完成时才提交第二个作业。然后，无论哪个作业先响应，都会提供结果值。