java - CompletableFuture join() 调用挂起主线程。个人的 future 永远不会完成

Question

我正在编写一个创建多个 (7) CompletableFutures 的函数。这些 future 基本上都做两件事:

1. 使用 SupplyAsync()，从某个数据库获取数据
2. 使用 thenAccept() 将此数据写入 CSV 文件

当所有 7 个 future 都完成工作后，我想继续进一步执行代码。因此，我使用 allOf()，然后对 allOf() 返回的 Void CompletableFuture 调用 join()。

问题是，即使在所有 future 都已执行之后(我可以看到生成的 CSV)， join() 调用仍然卡住，并且进一步的代码执行将永远被阻止。

我尝试过以下操作:

1. 在每个 future 之后调用 join() 逐一等待每个 future。这是可行的，但是是以并发为代价的。我不想这样做。

2. 尝试使用带有超时的 get() 而不是 join()。但是，这总是会引发异常(因为 get 总是超时)，这是不希望的。

3. 看到这个 JDK 错误:https://bugs.openjdk.java.net/browse/JDK-8200347 。不确定这是否是类似的问题。

4. 尝试在没有 join() 或 get() 的情况下运行，这不会保留线程执行，并且再次是不可取的。

创建所有 future 的主要函数。

public CustomResponse process() {
        CustomResponse msgResponse = new CustomResponse();
        try {
            // 1. DbCall 1
            CompletableFuture<Void> f1 = dataHelper.fetchAndUploadCSV1();

            // 2. DbCall 2
            CompletableFuture<Void> f2 = dataHelper.fetchAndUploadCSV2();


            // 3. DbCall 3
            CompletableFuture<Void> f3 = dataHelper.fetchAndUploadCSV3();


            // 4. DbCall 4
            CompletableFuture<Void> f4 = dataHelper.fetchAndUploadCSV4();


            // 5. DbCall 5
            CompletableFuture<Void> f5 = dataHelper.fetchAndUploadCSV5();


            // 6. DbCall 6
            CompletableFuture<Void> f6 = dataHelper.fetchAndUploadCSV6();


            // 7. DbCall 7
            CompletableFuture<Void> f7 = dataHelper.fetchAndUploadCSV7();


            CompletableFuture<Void>[] fAll = new CompletableFuture[] {f1, f2, f3, f4, f5, f6, f7};


            CompletableFuture.allOf(fAll).join();
            msgResponse.setProcessed(true);
            msgResponse.setMessageStatus("message");
        } catch (Exception e) {
    msgResponse.setMessageStatus(ERROR);
            msgResponse.setErrorMessage("error");
        }
        return msgResponse;
    }

每个 fetchAndUploadCSV() 函数如下所示:

public CompletableFuture<Void> fetchAndUploadCSV1() {
        return CompletableFuture.supplyAsync(() -> {
            try {
                return someService().getAllData1();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }).thenAccept(results -> {
            try {
                if (results.size() > 0) {
                    csvWriter.uploadAsCsv(results);
                }
                else {
                    log.info(" No data found..");
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }

这就是 csvWriter.uploadAsCsv(results) 的样子 -

public <T> void uploadAsCsv(List<T> objectList) throws Exception {
        long objListSize = ((objectList==null) ? 0 : objectList.size());
        log.info("Action=Start, objectListSize=" + objListSize);
        ByteArrayInputStream inputStream = getCsvAsInputStream(objectList);
        Info fileInfo = someClient.uploadFile(inputStream);
        log.info("Action=Done, FileInfo=" + ((fileInfo==null ? null : fileInfo.getID())));
    }

我在这里使用 OpenCSV 将数据转换为 CSV 流。而且我总是可以看到最后一行日志。

预期结果:获取的所有数据、生成的 CSV 和 CustomResponse 均应按处理后返回，且不会出现任何错误消息。

实际结果:所有数据均已获取，CSV 已生成，主线程挂起。

Answer 1

您可以在每个创建的 CompletableFuture 上使用 join，而不会牺牲并发性:

public CustomResponse process() {
    CustomResponse msgResponse = new CustomResponse();

    List<CompletableFuture<Void>> futures = Arrays.asList(dataHelper.fetchAndUploadCSV1(),
            dataHelper.fetchAndUploadCSV2(),
            dataHelper.fetchAndUploadCSV3(),
            dataHelper.fetchAndUploadCSV4(),
            dataHelper.fetchAndUploadCSV5(),
            dataHelper.fetchAndUploadCSV6(),
            dataHelper.fetchAndUploadCSV7());

    return CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0]))
            .thenApply(v -> {
                msgResponse.setProcessed(true);
                msgResponse.setMessageStatus("message");
                return msgResponse;
            })
            .exceptionally(throwable -> {
                msgResponse.setMessageStatus("ERROR");
                msgResponse.setErrorMessage("error");
                return msgResponse;
            }).join();
}

allOf返回一个新的 CompletableFuture，当所有给定的 CompletableFuture 完成时，该新的 CompletableFuture 也完成。因此，当在 thenApply 中调用 join 时，它会立即返回。本质上，加入是发生在已经完成的 future 上。这样就消除了阻塞。另外，为了处理可能的异常，应该调用exceptionally。