CompletableFuture in loop：如何收集所有响应并处理错误

Question

我正在尝试循环调用 PUT 请求的 rest api。每个电话都是一个CompletableFuture。每个 api 调用都返回一个 RoomTypes.RoomType 类型的对象

• 我想收集响应（成功和错误） 响应）在不同的列表中。我该如何做到这一点？我确定我 不能使用allOf，因为如果有的话，它不会得到所有的结果 一次调用更新失败。

• 如何记录每次调用的错误/异常？

public void sendRequestsAsync(Map<Integer, List> map1) {
    List<CompletableFuture<Void>> completableFutures = new ArrayList<>(); //List to hold all the completable futures
    List<RoomTypes.RoomType> responses = new ArrayList<>(); //List for responses
    ExecutorService yourOwnExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    for (Map.Entry<Integer, List> entry :map1.entrySet()) { 
        CompletableFuture requestCompletableFuture = CompletableFuture
                .supplyAsync(
                        () -> 
            //API call which returns object of type RoomTypes.RoomType
            updateService.updateRoom(51,33,759,entry.getKey(),
                           new RoomTypes.RoomType(entry.getKey(),map2.get(entry.getKey()),
                                    entry.getValue())),
                    yourOwnExecutor
            )//Supply the task you wanna run, in your case http request
            .thenApply(responses::add);

    completableFutures.add(requestCompletableFuture);
}

Answer 1

您可以简单地使用allOf() 来获得一个在您完成所有初始期货时完成的未来（无论是否例外），然后使用Collectors.partitioningBy() 将它们分为成功和失败：

List<CompletableFuture<RoomTypes.RoomType>> completableFutures = new ArrayList<>(); //List to hold all the completable futures
ExecutorService yourOwnExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

for (Map.Entry<Integer, List> entry : map1.entrySet()) {
    CompletableFuture<RoomTypes.RoomType> requestCompletableFuture = CompletableFuture
            .supplyAsync(
                    () ->
                //API call which returns object of type RoomTypes.RoomType
                updateService.updateRoom(51, 33, 759, entry.getKey(),
                        new RoomTypes.RoomType(entry.getKey(), map2.get(entry.getKey()),
                                entry.getValue())),
                    yourOwnExecutor
            );

    completableFutures.add(requestCompletableFuture);
}

CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0]))
        // avoid throwing an exception in the join() call
        .exceptionally(ex -> null)
        .join();
Map<Boolean, List<CompletableFuture<RoomTypes.RoomType>>> result =
        completableFutures.stream()
                .collect(Collectors.partitioningBy(CompletableFuture::isCompletedExceptionally)));

生成的映射将包含一个带有true 的条目用于失败的期货，另一个带有false 键的条目用于成功的期货。然后，您可以检查这 2 个条目以采取相应措施。

请注意，与您的原始代码相比，有 2 个细微的变化：

• requestCompletableFuture 现在是 CompletableFuture<RoomTypes.RoomType>
• thenApply(responses::add) 和 responses 列表已删除

关于日志记录/异常处理，只需添加相关的requestCompletableFuture.handle() 以单独记录它们，但保留requestCompletableFuture 而不是handle() 产生的那个。

Answer 2

或者，也许您可​​以从不同的角度解决问题，而不是强制使用CompletableFuture，而是使用CompletionService。

CompletionService 的整体理念是，一旦给定未来的答案准备就绪，它就会被放入队列中，您可以从中消费结果。

替代方案 1：没有 CompletableFuture

CompletionService<String> cs = new ExecutorCompletionService<>(executor);

List<Future<String>> futures = new ArrayList<>();

futures.add(cs.submit(() -> "One"));
futures.add(cs.submit(() -> "Two"));
futures.add(cs.submit(() -> "Three"));
futures.add(cs.submit(() -> { throw new RuntimeException("Sucks to be four"); }));
futures.add(cs.submit(() -> "Five"));


List<String> successes = new ArrayList<>();
List<String> failures = new ArrayList<>();

while (futures.size() > 0) {
    Future<String> f = cs.poll();
    if (f != null) {
        futures.remove(f);
        try {
            //at this point the future is guaranteed to be solved
            //so there won't be any blocking here
            String value = f.get();
            successes.add(value);
        } catch (Exception e) {
            failures.add(e.getMessage());
        }
    }
}

System.out.println(successes); 
System.out.println(failures);

产量：

[One, Two, Three, Five]
[java.lang.RuntimeException: Sucks to be four]

备选方案 2：使用 CompletableFuture

但是，如果您真的需要处理 CompletableFuture，您也可以将它们提交到完成服务，只需将它们直接放入队列即可：

例如，以下变体具有相同的结果：

BlockingQueue<Future<String>> tasks = new ArrayBlockingQueue<>(5);
CompletionService<String> cs = new ExecutorCompletionService<>(executor, tasks);

List<Future<String>> futures = new ArrayList<>();

futures.add(CompletableFuture.supplyAsync(() -> "One"));
futures.add(CompletableFuture.supplyAsync(() -> "Two"));
futures.add(CompletableFuture.supplyAsync(() -> "Three"));
futures.add(CompletableFuture.supplyAsync(() -> { throw new RuntimeException("Sucks to be four"); }));
futures.add(cs.submit(() -> "Five"));

//places all futures in completion service queue
tasks.addAll(futures);

List<String> successes = new ArrayList<>();
List<String> failures = new ArrayList<>();

while (futures.size() > 0) {
    Future<String> f = cs.poll();
    if (f != null) {
        futures.remove(f);
        try {
            //at this point the future is guaranteed to be solved
            //so there won't be any blocking here
            String value = f.get();
            successes.add(value);
        } catch (Exception e) {
            failures.add(e.getMessage());
        }
    }
}

Answer 3

对于你想使用 For 循环的地方。这是一个可行的解决方案。 CompletableFuture.allOf() ->

您想下载一个网站的 100 个不同网页的内容。您可以按顺序执行此操作，但这将花费大量时间。因此，可以编写一个获取网页链接并返回 CompletableFuture 的函数：

CompletableFuture<String> downloadWebPage(String pageLink) {
return CompletableFuture.supplyAsync(() -> {
    // Code to download and return the web page's content
});
} 

循环调用前面的函数，我们使用的是JAVA 8

List<String> webPageLinks = Arrays.asList(...)  // A list of 100 web page links

// Download contents of all the web pages asynchronously
List<CompletableFuture<String>> pageContentFutures = webPageLinks.stream()
    .map(webPageLink -> downloadWebPage(webPageLink))
    .collect(Collectors.toList());


// Create a combined Future using allOf()
CompletableFuture<Void> allFutures = CompletableFuture.allOf(
    pageContentFutures.toArray(new CompletableFuture[pageContentFutures.size()])
);

CompletableFuture.allOf() 的问题在于它返回 CompletableFuture。但是我们可以通过编写几行额外的代码来获得所有包装的 CompletableFutures 的结果

// When all the Futures are completed, call `future.join()` to get their results and collect the results in a list -
CompletableFuture<List<String>> allPageContentsFuture = allFutures.thenApply(v -> {
return pageContentFutures.stream()
       .map(pageContentFuture -> pageContentFuture.join())
       .collect(Collectors.toList());
});

现在让我们计算包含我们关键字的网页数量 ->

// Count the number of web pages having the "CompletableFuture" keyword.
CompletableFuture<Long> countFuture = allPageContentsFuture.thenApply(pageContents -> {
 return pageContents.stream()
        .filter(pageContent -> pageContent.contains("CompletableFuture"))
        .count();
});

System.out.println("Number of Web Pages having CompletableFuture keyword - " + 
    countFuture.get());