返回包含CompletableFuture列表的CompletableFuture

Question

我正在尝试使对多个API的调用更快。

在下面的代码中，getFilteredEvents是当前同步的版本。我感觉map(x -> x.getFilteredEvents(eventResearch))操作将等待每个API的响应（内部使用RestTemplate.exchange()），然后传递到下一个API，以构建要返回的List<Event>。一种解决方案可能是在单独的线程上启动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操作，创建了一个CompletableFuture来自多个CompleteableFuture）。

不幸的是，使用Generic array creation时出现new CompletableFuture<List<Event>>[0]编译错误。

我在做什么错？

我感觉使用join方法确实可以收集所有答案，但是那将是对Service线程的阻塞操作，不是吗？ （如果我理解正确的话，这样做会破坏尝试将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