CompletableFuture join() 调用挂起主线程.个人期货永远不会完成

Question

我正在编写一个创建多个(7(个CompletableFutures的函数。这些期货中的每一个基本上都做两件事：

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

当所有 7 个期货都完成工作时，我想继续进一步执行代码。所以，我正在使用allOf((，然后在allOf((返回的Void CompletableFuture上调用一个join((。

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

我尝试了以下方法：

1. 等待每个未来，在每个未来之后调用 join((。这有效，但以并发为代价。我不想这样做。

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

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

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

创建所有期货的主函数。

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 和自定义响应都应返回为已处理，没有错误消息。

实际结果 ： 获取所有数据，生成 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，当所有给定的 CompletableFutures 完成时完成。因此，当join在thenApply中被调用时，它会立即返回。从本质上讲，加入发生在已经完成的期货上。这样就可以消除阻塞。此外，若要处理可能的异常，应调用exceptionally。