是否可以使用 Java 8 Streams API 进行异步处理？

asynchronous java java-8 java-stream

2022-09-01 08:58:48

我一直在Java 8中使用ComputitionStage/CompletableFuture来做异步处理，这很好。但是，有时我想要一个阶段来执行迭代器/项目流的异步处理，但似乎没有办法做到这一点。

具体来说，Stream.forEach（）具有在调用后处理所有项的语义。我想要同样的东西，但用一个完成阶段代替，例如：

CompletionStage<Void> done = stream.forEach(...);
done.thenRun(...);

如果 Stream 由异步流式处理结果支持，则这比在上面的代码本身中等待它完成要好得多。

是否有可能以某种方式使用当前的Java 8 API构建它？解决方法？

答案 1

据我所知，流 API 不支持异步事件处理。听起来你想要类似.NET的Reactive Extensions，它有一个名为RxJava的Java端口，由Netflix创建。

RxJava 支持许多与 Java 8 流相同的高级操作（如 map 和 filter），并且是异步的。

更新：现在有一个反应式流计划正在进行中，看起来JDK 9将通过Flow类至少包括对部分内容的支持。

答案 2

正如@KarolKrol所暗示的那样，您可以使用.CompletableFuture

有一个库构建在JDK8流之上，以方便处理称为cyclops-react的流。CompletableFuture

要编写流，您可以使用cyclops-react的Fluent promise ike API，也可以使用simple-react的Stages。