java - 如何从 akka 流源获取迭代器？

Question

我正在尝试创建一个可以通过类似 Iterator 之类的方式使用的流程。 。我正在实现一个公开类似迭代器的接口(interface)的库，因此这对我来说是最简单的使用方式。

到目前为止，我设计的图表本质上是 Source<Iterator<DataRow>> 。到目前为止我看到的一件事是将其展平为 Source<DataRow>然后使用 http://doc.akka.io/japi/akka/current/akka/stream/javadsl/StreamConverters.html#asJavaStream--接下来是 https://docs.oracle.com/javase/8/docs/api/java/util/stream/BaseStream.html#iterator--

但是考虑到可能会有很多行，我想知道避免扁平化步骤是否有意义(至少在 akka 流上下文中，我假设传递时每个元素会有一些小的开销)通过阶段)，或者是否有更直接的方法。

另外，我很好奇反压在创建的流中是如何工作的，尤其是子迭代器；它只缓冲一个元素吗？

Answer 1

展平步骤

展平 Source<Iterator<DataRow>>到 Source<DataRow>确实会增加一些开销，因为您必须使用 flatMapConcat 最终create a new GraphStage 。

但是，如果您有“许多”行，那么这个单独的阶段可能会派上用场，因为它将为展平步骤提供并发性。

背压

如果你看at the code的StreamConverters.asJavaStream你会看到有一个 QueueSink生成一个 Future 从 akka 流中提取下一个元素，然后执行 Await.result(nextElementFuture, Inf)等待 Future 完成，以便下一个元素可以转发到 java Stream。

回答你的问题:是的，子迭代器仅缓冲一个元素，但 QueueSink 有一个 Future，它也可能有下一个 DataRow 。因此，除了原始 akka Source 中正在进行的缓冲之外，javaStream 和 Iterator 可能还缓冲了 2 个元素。 .