如何在CompletableFutures中收集成功和错误？

Question

我想向Web服务并行发送多个请求。结果应该由成功错误收集，然后可以由调用者进一步分析。

public Map.Entry<Rsp, Errors> sendApiRequests(List<Req> reqs) {
    //will mostly remain null as errors won't occur frequently
    List<Rsp> errors = null;

    List<CompletableFuture<Rsp>> futures =
            reqs.stream()
                .map(req -> CompletableFuture.supplyAsync(() -> send(req))
                        .exceptionally(ex -> {
                            //TODO this fails, because list should be final for it.
                            //but don't want to instantiate as mostly will remain just null
                            if (errors == null) errors = new ArrayList<>();
                            errors.add(req);
                        }))
                .collect(Collectors.toList());

    //send api requests in parallel
    List<Rsp> responses = futures.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());

    //TODO how to collect the errors? each error should also provide the underlying Req that caused the failure.
    //pending requests should not be aborted if any throwns an exception
    return new SimpleEntry(responses, errors);
}

问题：我怎样才能收集所有响应，同时也收集在send（）方法中抛出的异常？

我的目标不是返回两个列表：一个包含所有成功的响应，另一个包含错误。

Answer 1

虽然这是一个老问题，但我需要类似的东西，并且能够从上面的评论和其他不太正确的互联网片段中拼凑出一个可行的解决方案。

方法和CompletableFutureCollector类将为每个请求返回一个响应或错误列表。这在Java11中实现，但应该适用于Java8。我建议将其调整为传入java.util.concurrent.Execator以控制并行度。例如，您可以像这样使用它：

    final List<CompletableFutureCollector.CollectorResult<Rsp>> results =
            sendApiRequests(List.of(new Req()));
    results.stream()
            .filter(CompletableFutureCollector.CollectorResult::hasError)
            .map(CompletableFutureCollector.CollectorResult::getError)
            .forEach(error -> {
                // Do something with errors
            });
    results.stream()
            .filter(CompletableFutureCollector.CollectorResult::hasResult)
            .map(CompletableFutureCollector.CollectorResult::getResult)
            .forEach(rsp -> {
                // Do something with responses
            });

    public List<CompletableFutureCollector.CollectorResult<Rsp>> sendApiRequests(List<Req> reqs) {
        // The actual send implementation could be anything that you'd like to do asynchronously  
        return CompletableFutureCollector.mapAsyncAndCollectResult(reqs, req -> send(req));
    }

    // ...

    private final static class CompletableFutureCollector {
        private CompletableFutureCollector() {
        }

        public static <X, T extends CompletableFuture<X>> Collector<T, ?, CompletableFuture<List<X>>> collectResult() {
            return Collectors.collectingAndThen(Collectors.toList(), joinResult());
        }

        private static <X, T extends CompletableFuture<X>> Function<List<T>, CompletableFuture<List<X>>> joinResult() {
            return futures -> allOf(futures)
                    .thenApply(v -> futures.stream()
                            .map(CompletableFuture::join)
                            .collect(Collectors.toList()));
        }

        private static <T extends CompletableFuture<?>> CompletableFuture<Void> allOf(final List<T> futures) {
            return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
        }

        public static <T, R> List<CompletableFutureCollector.CollectorResult<R>> mapAsyncAndCollectResult(
                final Collection<T> items, final Function<T, R> action) {
            return items.parallelStream()
                    .map(task -> CompletableFuture.supplyAsync(() -> action.apply(task)))
                    .map(CompletableFutureCollector.CollectorResult::handle)
                    .collect(CompletableFutureCollector.collectResult())
                    .join();
        }

        private final static class CollectorResult<R> {
            private final R result;
            private final Throwable error;

            private CollectorResult(final R result, final Throwable error) {
                this.result = result;
                this.error = error;
            }

            public boolean hasError() {
                return getError() != null;
            }

            public boolean hasResult() {
                return !hasError();
            }

            public R getResult() {
                return result;
            }

            public Throwable getError() {
                return error instanceof CompletionException ? error.getCause() : error;
            }

            public static <R> CompletableFuture<CompletableFutureCollector.CollectorResult<R>> handle(final CompletableFuture<R> future) {
                return future.handle(CompletableFutureCollector.CollectorResult::new);
            }
        }
    }

注意：我还没有完全测试这个解决方案，因为它是从一个有一些额外的快速故障消除逻辑的工作实现中改编的。