java - 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(查询); 返回一个List<Results>所以输出是List<List<Result>. 我想将所有结果展平并合并到一个列表中。如果我在收集之前使用 .flatMap(List::stream) 来展平，它会保持排序吗？

Answer 1

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

考虑显式传递一个ExecutortosupplyAsync以避免在ForkJoinPool.commonPool().

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

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

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