为什么 Source.fromIterator 需要 Function0[Iterator[T]] 而不是 Iterator[T] 作为参数？

Question

我不明白为什么Source.fromIterator的参数是Function0[Iterator[T]]而不是Iterator[T]。

这有实际原因吗？我们可以将签名改为 def fromIterator(iterator: => Iterator[T]) 吗？（避免做 Source.fromIterator( () => myIterator) ）

Answer 1

The iterator will be created anew for each materialization, which is the reason the method takes a function rather than an iterator directly.

Stream 阶段应该是可重复使用的，因此您可以实现多个阶段。但是，给定的迭代器（通常）只能被使用一次。如果 fromIterator 创建了一个引用现有迭代器的源（无论是通过名称还是引用传递），第二次尝试实现它可能会失败，因为底层迭代器将被耗尽。

为了解决这个问题，源需要能够实例化一个新的迭代器，因此 fromIterator 允许您提供必要的逻辑来作为供应商函数执行此操作。

这是我们不希望发生的事情的示例：

implicit val system = akka.actor.ActorSystem.create("test")
implicit val mat = akka.stream.ActorMaterializer(system)

val iter = Iterator.range(0, 2)
// pretend we pass the iterator directly...
val src = Source.fromIterator(() => iter)

Await.result(src.runForEach(println), 2.seconds)
// 0
// 1
// res0: akka.Done = Done

Await.result(src.runForEach(println), 2.seconds)
// res1: akka.Done = Done
// No results???

这很糟糕，因为 Source src 不可重复使用，因为它在后续运行中不会提供相同的输出。但是，如果我们懒惰地创建迭代器，它会起作用：

val iterFunc = () => Iterator.range(0, 2)
val src = Source.fromIterator(iterFunc)

Await.result(src.runForEach(println), 2.seconds)
// 0
// 1
// res0: akka.Done = Done

Await.result(src.runForEach(println), 2.seconds)
// 0
// 1
// res1: akka.Done = Done

为什么 Source.fromIterator 需要 Function0[Iterator[T]] 而不是 Iterator[T] 作为参数？

Why Source.fromIterator expects a Function0[Iterator[T]] as a parameter instead of Iterator[T]?

scala

akka-stream