java - 使用 CompletableFuture 时如何避免编译器警告关闭 OutputStream？

Question

我使用 CompletableFuture 进行异步操作，下载文件并通过 OutputStream 保存其内容。下面的代码可以工作，但编译器会警告我使用 try-with-resources 或在 finally-clause 中关闭 OutputStream，尽管它是在 Future 在 whenComplete 中完成后关闭的。

代码:

final OutputStream outputStream = Files.newOutputStream(file.toPath());
final String url = "https://example.com/some-download.zip";
final CompletionStage<WSResponse> futureResponse = this.client
        .url(url)
        .setMethod("GET")
        .stream();

futureResponse.thenCompose(res -> {
    downloadTask.setTotalBytes(res);
    Source<ByteString, ?> responseBody = res.getBodyAsSource();

    Sink<ByteString, CompletionStage<akka.Done>> outputWriter =
            Sink.foreach(bytes -> {
                downloadTask.addReceivedBytes(bytes.size());
                System.out.println(downloadTask.getProgressAsString());
                outputStream.write(bytes.toArray());
            });

    return responseBody.runWith(outputWriter, this.materializer);
}).whenComplete((res, error) -> {
    try {
        outputStream.close();
    } catch (final IOException e) {
        e.printStackTrace();
    }
});

警告:

问题:

当我使用 try-with-resources 时，程序会在将任何内容写入文件之前关闭 OutPutStream，因为 CompletableFuture 的异步性质不会阻塞。

那么有没有办法在 CompletionStage 中声明 OutputStream 并将其向下传递？

Answer 1

@AndyTurner 在评论中的建议是正确的。我只需稍微调整一下代码即可使其工作。

我没有意识到responseBody.runWith()返回另一个CompletionStage，因此使用try-with-resources或finally关闭了thenComponse block 中的OutputStream，因此返回的CompletionStage(来自responseBody.runWith() )无法再写入。这导致了错误。因此，我们只需使用 .toCompletableFuture().get() 在同一 block 中“同步”处理 responseBody.runWith() 即可。这没有问题，因为 block 本身在另一个线程中运行，即它保持异步。由于我们不再返回任何内容，因此我们还需要使用 thenAccept，它需要一个 Consumer，而不是 thenCompose，它需要一个 Function。

final String url = "https://example.com/some-download.zip";
final CompletionStage<WSResponse> futureResponse = this.client
        .url(url)
        .setMethod("GET")
        .stream();

futureResponse
        .thenAccept(res -> {
            try (OutputStream outputStream = Files.newOutputStream(file.toPath())) {
                downloadTask.setTotalBytes(res);
                Source<ByteString, ?> responseBody = res.getBodyAsSource();

                Sink<ByteString, CompletionStage<akka.Done>> outputWriter =
                        Sink.foreach(bytes -> {
                            downloadTask.addReceivedBytes(bytes.size());
                            System.out.println(downloadTask.getProgressAsString());
                            outputStream.write(bytes.toArray());
                        });
                responseBody.runWith(outputWriter, this.materializer).toCompletableFuture().get();
            } catch (IOException | InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });