CompletableFuture：运行期货列表的正确方法，等待结果并处理异常

Question

我有一个遗留代码，它有十几个数据库调用来填充报告，这需要大量时间，我试图使用 CompletableFuture 来减少它。

我有些怀疑我做事是否正确并且没有过度使用这项技术。

我的代码现在看起来像这样：

开始异步填充文档部分，每个方法内部都有许多数据库调用

CompletableFuture section1Future = CompletableFuture.supplyAsync(() -> populateSection1(arguments));
CompletableFuture section2Future = CompletableFuture.supplyAsync(() -> populateSection2(arguments));
    ...
CompletableFuture section1oFuture = CompletableFuture.supplyAsync(() -> populateSection10(arguments));

然后我在 arrayList 中按特定顺序安排期货并加入所有期货以确保我的代码仅在所有期货完成后才会运行进一步。

List<CompletableFuture> futures = Arrays.asList(
            section1Future,
            section2Future, ...
            section10Future);

List<Object> futureResults = futures.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());

然后我用它的片段填充 PDF 文档本身

Optional.ofNullable((PdfPTable) futureResults.get(0)).ifPresent(el -> populatePdfElement(document, el));
Optional.ofNullable((PdfPTable) futureResults.get(1)).ifPresent(el -> populatePdfElement(document, el));
    ...
Optional.ofNullable((PdfPTable) futureResults.get(10)).ifPresent(el -> populatePdfElement(document, el));

return 文档

我的顾虑是：

1) 以这种方式创建和实例化许多 Completable Future 是否可以？在arrayList中按要求的顺序排序，加入它们确保它们都完成，然后将它们转换为特定对象得到结果？

2) 运行不指定执行器服务而是依赖公共 ForkJoinPool 可以吗？但是这段代码运行s 在 web 容器中，所以可能为了使用 JTA 我需要通过 JNDI 使用容器提供的线程池执行器？

3) 如果这段代码包含在 try-catch 中，我应该能够在主线程中捕获 CompletionException，对吧？或者为了做到这一点，我应该像下面这样声明每个特性：

CompletableFuture.supplyAsync(() -> populateSection1(arguments))
    .exceptionally (ex -> {
                    throw new RuntimeException(ex.getCause());
        });

4) 是否有可能过度使用 CompletableFutures 使它们本身成为性能瓶颈？像许多期货一样等待一位执行者开始运行ning？如何避免这种情况？使用容器提供的执行器服务？如果是，有人可以告诉我一些关于如何在考虑处理器和内存量的情况下正确配置执行程序服务的最佳实践吗？

5) 内存影响。我在并行线程中读到 OOME 可能存在问题，因为创建了许多对象并收集了垃圾。是否有关于如何计算应用程序所需的正确内存量的最佳实践？

Answer 1

总的来说，方法没有错，但也有需要改进的地方。

最值得注意的是，您不应该使用 原始类型，例如 CompletableFuture.

当 populateSection… return 是 PdfPTable 时，您应该在整个代码中始终使用 use CompletableFuture<PdfPTable>。

即

CompletableFuture<PdfPTable> section1Future = CompletableFuture.supplyAsync(()  -> populateSection1(arguments));
CompletableFuture<PdfPTable> section2Future = CompletableFuture.supplyAsync(()  -> populateSection2(arguments));
    ...
CompletableFuture<PdfPTable> section10Future = CompletableFuture.supplyAsync(() -> populateSection10(arguments));

即使这些方法没有声明您假设总是在运行时 returned 的 return 类型，您也应该在这个早期阶段插入类型转换：

CompletableFuture<PdfPTable> section1Future = CompletableFuture.supplyAsync(()  -> (PdfPTable)populateSection1(arguments));
CompletableFuture<PdfPTable> section2Future = CompletableFuture.supplyAsync(()  -> (PdfPTable)populateSection2(arguments));
    ...
CompletableFuture<PdfPTable> section10Future = CompletableFuture.supplyAsync(() -> (PdfPTable)populateSection10(arguments));

然后，您可以使用

Stream.of(section1Future, section2Future, ..., section10Future)
    .map(CompletableFuture::join)
    .filter(Objects::nonNull)
    .forEachOrdered(el -> populatePdfElement(document, el));

通过不使用原始类型，您已经获得了所需的结果类型，您可以在这个流操作中执行第 3 步的操作，即过滤和执行最终操作。

如果您还需要列表，您可以使用

List<PdfPTable> results = Stream.of(section1Future, section2Future, ..., section10Future)
    .map(CompletableFuture::join)
    .filter(Objects::nonNull)
    .collect(Collectors.toList());

results.forEach(el -> populatePdfElement(document, el));

也就是说，并行度取决于用于操作的线程池（指定为 supplyAsync）。当你不指定执行者时，你会得到并行流使用的默认 Fork/Join 池，所以在这种特定情况下，你得到与

相同的结果要简单得多

List<PdfPTable> results = Stream.<Supplier<PdfPTable>>.of(
    ()  -> populateSection1(arguments),
    ()  -> populateSection2(arguments));
    ...
    () -> populateSection10(arguments)))
    .parallel()
    .map(Supplier::get)
    .filter(Objects::nonNull)
    .forEachOrdered(el -> populatePdfElement(document, el));

或

List<PdfPTable> results = Stream.<Supplier<PdfPTable>>.of(
    ()  -> populateSection1(arguments),
    ()  -> populateSection2(arguments));
    ...
    () -> populateSection10(arguments)))
    .parallel()
    .map(Supplier::get)
    .filter(Objects::nonNull)
    .collect(Collectors.toList());

results.forEach(el -> populatePdfElement(document, el));

虽然这两种变体都确保 populatePdfElement 将以正确的顺序调用，并且一次调用一个，但只有后者会执行来自启动线程的所有调用。

关于异常处理，当您调用 CompletableFuture::join 时，您会收到供应商抛出的任何异常，这些异常包含在 CompletionException 中。链接 .exceptionally (ex -> { throw new RuntimeException(ex.getCause()); }); 这样的东西是没有意义的，当你调用 CompletableFuture::join.

时，新的 RuntimeException 也会被包裹在 CompletionException 中

在 Stream 变体中，您将获得没有包装器的异常。由于 Supplier 不允许检查异常，因此只有 RuntimeException 或 Error 的子类型是可能的。

其他问题对于问答来说太宽泛了。

CompletableFuture：运行期货列表的正确方法，等待结果并处理异常

CompletableFuture: proper way to run a list of futures, wait for result and handle exception

java

design-patterns

asynchronous

java-8