Return 包含 CompletableFuture 列表的 CompletableFuture

Question

我正在努力使对多个 API 的调用更快。

在下面的代码中，getFilteredEvents 是当前同步的版本。我感觉 map(x -> x.getFilteredEvents(eventResearch)) 操作将等待每个 API 的响应（它在内部使用 RestTemplate.exchange()），然后再传递到下一个以构建我认为的 List<Event>想要return。一个解决方案可能是在单独的线程上启动 map 调用，但我想尝试 CompletableFuture API.

因此，getFilteredEventsFaster是我努力改善响应时间的结果。

@Service
public class EventsResearchService {

    @Autowired
    private List<UniformEventsResearchApi> eventsResearchApis;

    // this works, but I'm trying to improve it
    public EventResearchResponse getFilteredEvents(EventResearch eventResearch) {
        List<Event> eventsList = eventsResearchApis
                .stream()
                .map(x -> x.getFilteredEvents(eventResearch))
                .flatMap(List::stream)
                .collect(Collectors.toList());

        return extractResponse(eventResearch, eventsList);
    }

    // this doesn't work yet: what is wrong?
    public CompletableFuture<List<Event>> getFilteredEventsFaster(EventResearch eventResearch) {
        List<CompletableFuture<List<Event>>> futureEventsList = eventsResearchApis
                .parallelStream()
                .map(x -> CompletableFuture.supplyAsync(() -> x.getFilteredEvents(eventResearch)))
                .collect(Collectors.toList());

        return CompletableFuture.allOf(futureEventsList.toArray(new CompletableFuture<List<Event>>[0]));
    }
}

我的理解是我想将 CompletableFuture<List<Event>> 发送回我的前端，而不是 List<CompletableFuture<List<Event>>>，因此 CompletableFuture.allOf() 调用（如果我理解正确的话，类似于 flatmap 操作，从多个 CompleteableFutures.

创建一个 CompletableFuture

不幸的是，我在使用 new CompletableFuture<List<Event>>[0].

时遇到 Generic array creation 编译错误

我做错了什么？

我感觉使用join 方法确实可以让我收集所有的答案，但那将是对我的服务线程的阻塞操作，不是吗？（如果我理解正确的话，这会破坏尝试 return 一个 CompletableFuture 到我的前端的目的。）

Answer 1

以下代码片段显示了使用 listOfFutures.stream().map(CompletableFuture::join) 收集 allOF 的结果。我从 this page 中获取了这个例子，它声明它不会等待每个 Future 完成。

class Test {

    public static void main(String[] args) throws Exception {

        long millisBefore = System.currentTimeMillis();

        List<String> strings = Arrays.asList("1","2", "3", "4", "5", "6", "7", "8");
        List<CompletableFuture<String>> listOfFutures = strings.stream().map(Test::downloadWebPage).collect(toList());
        CompletableFuture<List<String>> futureOfList = CompletableFuture
                .allOf(listOfFutures.toArray(new CompletableFuture[0]))
                .thenApply(v ->  listOfFutures.stream().map(CompletableFuture::join).collect(toList()));

        System.out.println(futureOfList.get()); // blocks here
        System.out.printf("time taken : %.4fs\n", (System.currentTimeMillis() - millisBefore)/1000d);
    }

    private static CompletableFuture<String> downloadWebPage(String webPageLink) {
        return CompletableFuture.supplyAsync( () ->{
            try { TimeUnit.SECONDS.sleep(4); }
            catch (Exception io){ throw new RuntimeException(io); }
            finally { return "downloaded : "+ webPageLink; }
            });
    }

}

由于效率似乎是这里的问题，我提供了一个虚拟基准测试来证明它不需要 32 秒来执行。

输出：

[downloaded : 1, downloaded : 2, downloaded : 3, downloaded : 4, downloaded : 5, downloaded : 6, downloaded : 7, downloaded : 8]
time taken : 8.0630s

从原始问题海报编辑

感谢这个答案，并通过使用 this website（讨论与 allOf 相关的异常处理），我想出了这个完整的版本：

    public CompletableFuture<List<Event>> getFilteredEventsFaster(EventResearch eventResearch) {

        /* Collecting the list of all the async requests that build a List<Event>. */
        List<CompletableFuture<List<Event>>> completableFutures = eventsResearchApis.stream()
                .map(api -> getFilteredEventsAsync(api, eventResearch))
                .collect(Collectors.toList());

        /* Creating a single Future that contains all the Futures we just created ("flatmap"). */
        CompletableFuture<Void> allFutures =CompletableFuture.allOf(completableFutures
                .toArray(new CompletableFuture[eventsResearchApis.size()]));

        /* When all the Futures have completed, we join them to create merged List<Event>. */
        CompletableFuture<List<Event>> allCompletableFutures = allFutures
                .thenApply(future -> completableFutures.stream()
                            .map(CompletableFuture::join)
                            .flatMap(List::stream) // creating a List<Event> from List<List<Event>>
                            .collect(Collectors.toList())
                );

        return allCompletableFutures;
    }

    private CompletableFuture<List<Event>> getFilteredEventsAsync(UniformEventsResearchApi api,
            EventResearch eventResearch) {
        /* Manage the Exceptions here to ensure the wrapping Future returns the other calls. */
        return CompletableFuture.supplyAsync(() -> api.getFilteredEvents(eventResearch))
                .exceptionally(ex -> {
                    LOGGER.error("Extraction of events from API went wrong: ", ex);
                    return Collections.emptyList(); // gets managed in the wrapping Future
                });
    }

Return 包含 CompletableFuture 列表的 CompletableFuture

Return a CompletableFuture containing a list of CompletableFutures

java

spring

spring-mvc

spring-boot

completable-future

从原始问题海报编辑