Java 从多次调用中收集 CompletableFuture 的结果

Question

我必须运行多个外部调用操作，然后以列表的形式获取结果。我决定使用 CompletableFuture api，而我准备的代码非常恶心:

示例:

public class Main {
    public static void main(String[] args) {
        String prefix = "collection_";

        List<CompletableFuture<User>> usersResult = IntStream.range(1, 10)
                .boxed()
                .map(num -> prefix.concat("" + num))
                .map(name -> CompletableFuture.supplyAsync(
                        () -> callApi(name)))
                .collect(Collectors.toList());

        try {
            CompletableFuture.allOf(usersResult.toArray(new CompletableFuture[usersResult.size()])).get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        List<User> users = usersResult //the result I need
                .stream()
                .map(userCompletableFuture -> {
                    try {
                        return userCompletableFuture.get();
                    } catch (InterruptedException | ExecutionException e) {
                        e.printStackTrace();
                    }
                    return null;
                })
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

    private static User callApi(String collection) {
        return new User(); //potentially time-consuming operation
    }
}

我有以下问题:

1. 我能否以某种方式避免在流中重复 try-catch block (将 CompletableFuture 映射到 User)？
2. 这段代码可以减少顺序吗(如何避免等待所有 future 完成？)

3. 这样做可以吗(所有的 future 都会在流中解析吗？):

   public class Main {
       public static void main(String[] args) {
           String prefix = "collection_";
   
           List<User> usersResult = IntStream.range(1, 10)
                   .boxed()
                   .map(num -> prefix.concat("" + num))
                   .map(name -> CompletableFuture.supplyAsync(
                           () -> callApi(name)))
                   .filter(Objects::nonNull)
                   .map(userCompletableFuture -> {
                       try {
                           return userCompletableFuture.get();
                       } catch (InterruptedException | ExecutionException e) {
                           e.printStackTrace();
                       }
                       return null;
                   })
                   .collect(Collectors.toList());
       }
   
       private static User callApi(String collection) {
           return new User(); //potentially time-consuming operation
       }
   }
   

Answer 1

对于 1.，您可以完全跳过 allOf().get()调用，因为无论如何你都在一一等待所有的 future。 ¹

对于 2.，您可以简化 try-catch通过执行以下操作:

• 使用exceptionally()将来直接处理异常；
• 使用join()而不是get()以避免检查异常(并且您知道不可能出现异常)。

对于 3.，你不能真正减少它的顺序，因为你至少需要执行以下步骤:创建所有 future，然后处理它们的结果。

如果您在一个流中完成所有操作，它将创建每个 future ，然后立即等待它，然后再创建下一个 - 因此您将失去并行性。您可以使用并行流，但使用 CompletableFuture 不会有太大好处。 s。

所以最终的代码是:

List<CompletableFuture<User>> usersResult = IntStream.range(1, 10)
        .boxed()
        .map(num -> prefix.concat("" + num))
        .map(name -> CompletableFuture.supplyAsync(() -> callApi(name))
            .exceptionally(e -> {
                e.printStackTrace();
                return null;
            }))
        .collect(Collectors.toList());

List<User> users = usersResult
        .stream()
        .map(CompletableFuture::join)
        .filter(Objects::nonNull)
        .collect(Collectors.toList());

¹ 请注意 allOf()如果您希望结果为CompletableFuture<List<User>>，则仍然需要调用以及，例如

final CompletableFuture<List<User>> result =
        CompletableFuture.allOf(usersResult.stream().toArray(CompletableFuture[]::new))
                .thenApply(__ -> usersResult
                        .stream()
                        .map(CompletableFuture::join)
                        .filter(Objects::nonNull)
                        .collect(Collectors.toList()));