java - CompletableFutures 并根据内部值进行过滤

Question

我现在有点困惑，所以我有一个方法应该返回 CompletableFuture<List<A>>

方法内部是:

CompletableFuture<List<String>> toReturn = asyncCall().thenApply(....)
.thenCompose(listOfStuff -> convertToList(listOfStuff.stream().map(
     key -> asyncCall2(key)
        .thenApply(optionalValue -> optionalValue.orElse(null))
).collect(Collectors.toList()));

和convertToList()只需加入 future 即可兑换CompletableFuture<List<ComputableFuture<A>>>进入CompletableFuture<List<A>>

基本上我的目的是过滤 optionalValue.orElse(null) 中出现的空值在将所有内容收集到最后一行中列出之前，很容易进行过滤，但是如果我在 .collect 之前使用它它正在 CompletableFutures 上运行

我怀疑我可以在我的代码中进行很多重组。

编辑:

private<T> CompletableFuture<List<T>> convertToList(List<CompletableFuture<T>> toConvert) {
    return CompletableFuture.allOf(toConvert.toArray(new CompletableFuture[toConvert.size()]))
            .thenApply(v -> toConvert.stream()
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList())
            );
}

Answer 1

最好的方法可能是更改 convertToList() 以便它不返回列表的 future ，而是返回流:

private <T> CompletableFuture<Stream<T>> convertToFutureOfStream(List<CompletableFuture<T>> toConvert) {
    return CompletableFuture.allOf(toConvert.stream().toArray(CompletableFuture[]::new))
            .thenApply(
                    v -> toConvert.stream()
                            .map(CompletableFuture::join)
            );
}

这将更具可重用性，因为该方法将允许更好的链接，并且不会强制调用者使用列表，同时仍然允许通过简单的收集轻松获取列表。

然后您可以简单地过滤该流以删除空选项:

CompletableFuture<List<String>> toReturn = asyncCall()
    .thenCompose(listOfStuff -> convertToFutureOfStream(
            listOfStuff.stream()
                    .map(this::asyncCall2)
                    .collect(Collectors.toList())
        )
        .thenApply(stream ->
                stream.filter(Optional::isPresent)
                        .map(Optional::get)
                        .collect(Collectors.toList())
        )

    );

您甚至可以通过更改 convertToFutureOfStream() 以将流作为参数来进一步改进:

private <T> CompletableFuture<Stream<T>> convertToFutureOfStream(Stream<CompletableFuture<T>> stream) {
    CompletableFuture<T>[] futures = stream.toArray(CompletableFuture[]::new);
    return CompletableFuture.allOf(futures)
            .thenApply(v -> Arrays.stream(futures).map(CompletableFuture::join));
}

(不幸的是，由于泛型类型数组，这会引发未经检查的赋值警告)

然后给出

CompletableFuture<List<String>> toReturn = asyncCall()
    .thenCompose(listOfStuff -> convertToFutureOfStream(
                listOfStuff.stream().map(this::asyncCall2)
            )
        .thenApply(stream ->
                stream.filter(Optional::isPresent)
                        .map(Optional::get)
                        .collect(Collectors.toList())
        )

    );