CompletableFuture join（）调用挂起主线程。个人期货从未完成

Question

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

1. 使用supplyAsync（）从某些数据库获取数据
2. 使用thenAccept（）将此数据写入CSV文件

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

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

我试过以下几件事：

>

逐个等待每个将来，在每个将来之后调用join（）。这是可行的，但代价是并发性。我不想这么做。

尝试使用带有超时的get（）而不是join（）。但是，这总是导致抛出异常（因为get总是超时），这是不希望的。

看到了这个JDK bug:https://bugs.openjdk.java.net/browse/jdk-8200347。不确定这是否是一个类似的问题。

尝试在没有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和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，当所有给定的CompletableFutures完成时，该值将完成。因此，当thenapply中调用join时，它会立即返回。本质上，加入正在发生在已经完成的期货上。这样就消除了阻塞。此外，为了处理可能的异常，应该调用exceptionary。