Java 8 维护流顺序 CompletableFuture::join

Question

我有一个异步执行的查询输入流。我想确保当我使用 Completablefuture::join 时，这些要求的结果按输入查询流的顺序收集。

我的代码是这样的：

queries.stream()
     .map(query -> CompletableFuture.supplyAsync(() -> {
                    try {
                        return SQLQueryEngine.execute(query);
                    } catch (InternalErrorException e) {
                        throw new RuntimeException(e);
                    }
     }))
     .map(CompletableFuture::join)
     .collect(Collectors.toList());

SQLQueryEngine.execute(查询); returns 一个 List<Results> 所以输出是 List<List<Result>。我想将所有结果展平并合并到一个列表中。如果我在收集之前使用 .flatMap(List::stream) 来展平，它会保持顺序吗？

Answer 1

您的意思可能是 .flatMap 是的，它会保留顺序。

考虑将 Executor 显式传递给 supplyAsync 以避免在 ForkJoinPool.commonPool().

中安排您的 IO 绑定 sql 查询

作为@Ruben ，您在提交任务后立即加入当前线程中的每个任务，在提交下一个查询之前，这可能是一个错误。您应该先提交所有查询，然后才开始加入。

你可以这样做（静态导入toList）：

queries.stream()
    .map(query -> CompletableFuture.supplyAsync(...))
    .collect(toList())
    .stream()
    .map(CompletableFuture::join)
    .collect(toList());

Java 8 维护流顺序 CompletableFuture::join

Java 8 maintain stream order with CompletableFuture::join

java

java-8

java-stream

completable-future