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

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）以这种方式创建和实例化许多可完成的期货可以吗？在arraylist中按照所需的顺序排列它们，将它们连接起来以确保它们全部完成，然后将它们转换到特定对象中得到结果？

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

3）如果这段代码被包围在try-catch中，我应该能够在主线程中捕获completionexception，对吗？或者为了做到这一点，我应该声明每个特性，如下所示：

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

4）是否有可能过度使用CompletableFuture，从而使它们本身成为性能瓶颈？就像许多futures等待一个执行器开始运行一样？如何避免这种情况？使用容器提供的执行器服务？如果是，请有人给我指出一些最佳实践，如何正确配置执行器服务，考虑到处理器和内存数量？

5）内存影响。我在并行线程中读到，当许多对象被创建和垃圾收集时，可能会有问题。有没有一个最佳实践来计算应用程序所需的正确内存数量？

Answer 1

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

最值得注意的是，您不应该使用原始类型，如completablefuture。

当populatesection…返回pdfptable时，您应该在整个代码中一致地使用usecompletablefuture 。

即。

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

通过不使用原始类型，您已经获得了所需的结果类型，并且可以在该流操作中执行第三步的操作，即筛选和执行最终操作。

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

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

关于异常处理，当您调用CompletableFuture::Join时，您将得到包装在CompletionException中的供应商引发的任何异常。在例外情况下（ex->{throw new RuntimeException（ex.getcause（））；}）；没有意义，当您调用completablefuture::join时，新的RuntimeException也将包装在completionexception中。

在Stream变体中，您将在没有包装器的情况下获得异常。由于supplier不允许检查异常，因此只能使用runtimeexception或error的子类型。

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