java流: custom collector

Question

这是我的流:

Stream<CompletableFuture<String>> futureStream = IntStream
    .iterate(1, n -> n < resultSet.getTotalCount() / pageSize, n -> n++)
    .mapToObj(pageNumber -> this.buildCompletableFutureofResultSetType(oid, pageNumber, pageSize));

CompletableFuture有一个 allOf :

public static CompletableFuture<Object> allOf(CompletableFuture<?>... cfs);

我看过 allOf方法代码:

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
    return andTree(cfs, 0, cfs.length - 1);
}

我知道我可以先将它们全部收集到 List<CompletableFuture>然后将其传递给 allOf .
但是，我想链接这个 collector process进入流功能链。
有任何想法吗？

Answer 1

However, I'd like to chain this collector process into stream functional chain.

不，你没有。 andTree方法试图建立一个平衡的树。尝试自己构建平衡树并非易事，甚至可能是不可能的( CompletableFuture 可以构建平衡树的事实是一个实现细节，因为 API 没有公开任何这个，因此你不能使用它；这些都是私有(private)/包私有(private)方法)。
与此同时， Collection 家是相当复杂的概念。他们需要能够即时平衡并填补潜在的 并行其排序和排序属性可变的传入数据流。因此，编写自己的收集器是可能的，但这样做很容易使自己陷入困境。
所以，如果你想“直接”将你的 future 流链接成一个单一的 future ，当流中的所有元素都复杂化时，它会非常复杂并且会导致不平衡的树。很有可能 减 比引入通过中间列表的开销更有效。
我们在谈论多少个 future ？如果答案是“小于 10000”，则该中间列表的成本实际上为零。如果答案是“超过 10000”，那么该列表的成本可能是有问题的，但是对于这些 future 而言，拥有不平衡树的问题要大得多。无论哪种方式，直接收集流都是错误的答案。
因此，你想要什么:

streamOfFutures.collect(Collectors.collectingAndThen(Collectors.toList(), list -> CompletableFuture.allOf(list.toArray(CompletableFuture[]::new))));

// or possibly, as it's much cleaner and shorter...
// Remember, functional style / streams are a tool, not some sort of magic
// incantation that guarantees elegant code!

CompletableFuture.allOf(streamOfFutures.toArray(CompletableFuture[]::new));