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

asynchronous multithreading java java-8 completable-future

2022-09-02 19:58:38

我正在尝试在循环中调用 rest api 以获取请求。每个调用都是一个 .每个 api 调用返回一个对象类型PUTCompletableFutureRoomTypes.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);
}

答案 1

您可以简单地使用，以获得在所有初始期货完成（特殊或非特殊情况下）完成的未来，然后使用 Collectors.partitioningBy（）将它们拆分为成功和失败：allOf()

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)));

生成的映射将包含一个用于失败期货的条目，另一个条目包含用于成功期货的键。然后，您可以检查 2 个条目以采取相应的操作。truefalse

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

requestCompletableFuture现在是CompletableFuture<RoomTypes.RoomType>
thenApply(responses::add)并且该列表已被删除responses

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

答案 2

或者，也许您可以从不同的角度处理问题，而不是强制使用，而是使用完井服务。CompletableFuture

的整个想法是，一旦给定未来的答案准备就绪，它就会被放置在队列中，您可以从中使用结果。CompletionService

备选方案1：没有可满足的将来

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

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

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());
        }
    }
}