java - CompletableFuture : proper way to run a list of futures, 等待结果并处理异常

Question

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

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

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

1. 开始异步填充文档部分，在每个方法中调用许多数据库

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

2. 然后我在 arrayList 中按特定顺序安排 futures 并加入所有 futures 以确保只有当所有 futures 完成时我的代码才会进一步运行。

   List<CompletableFuture> futures = Arrays.asList(
               section1Future,
               section2Future, ...
               section10Future);
   
   List<Object> futureResults = futures.stream()
               .map(CompletableFuture::join)
               .collect(Collectors.toList());
   

3. 然后我用它的片段填充 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));
   

   返回文件

我的担忧是:

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

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

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

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

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

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

Answer 1

这种做法总体上没有错，但也有需要改进的地方。

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

当 populateSection…返回 PdfPTable , 你应该使用 CompletableFuture<PdfPTable>在整个代码中保持一致。

即

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

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

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将以正确的顺序调用，一次调用一个，只有后者会执行来自发起线程的所有调用。

关于异常处理，您会收到供应商抛出的任何异常，包裹在 CompletionException 中当你调用CompletableFuture::join .链接类似 .exceptionally (ex -> { throw new RuntimeException(ex.getCause()); }); 的内容没有意义，新的 RuntimeException也将被包装在 CompletionException 中当你调用CompletableFuture::join .

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

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