如何在Scala中复制迭代器？

Question

关于重复

这不是如何克隆的重复迭代器？

请不要盲目地关闭这个问题，所有所谓重复的答案都不起作用。 OP 负责另一个问题，显然，答案适合他的问题，但不适合我的问题。

并不是每个类似的问题都是重复的，SE上有“扩展问题”这样的功能，唯一的方法是就同一主题再次询问以获得不同的、有效的答案。

问题

我有迭代器。我想获得它的副本（副本），这样我就可以完全独立地处理原件和副本。

重要

通过反射或序列化进行复制是不行的（性能损失）。

示例

var list = List(1,2,3,4,5)
var it1 = list.iterator
it1.next()

var it2 = it1   // (*)
it2.next()

println(it1.next())

这将简单地引用it1，因此当更改it1时，it2也会更改，反之亦然。

上面的例子使用List，我目前正在努力使用HashMap，但问题是一般性的——只是迭代器。

方法#1

如果你编辑行（*）并写入：（

var it2 = it1.toList.iterator

这是建议作为链接问题中的解决方案）执行程序时抛出异常。

方法＃2

“你拿着清单然后......”。不，我不。我没有列表，我有迭代器。一般来说，我对迭代器背后的集合一无所知，我唯一拥有的就是迭代器。我必须“分叉”它。

Answer 1

您无法在不破坏迭代器的情况下复制它。迭代器的约定是它只能被遍历一次。

您链接到的问题显示了如何用您已销毁的一份来换取两份副本。您不能继续使用原始版本，但现在可以独立运行两个新副本。

Answer 2

创建一个 List 迭代器非常容易，您可以复制它而不破坏它：这基本上是从 List 复制的 iterator 方法的定义添加了 fork 方法的源代码：

class ForkableIterator[A] (list: List[A]) extends Iterator[A] {
    var these = list
    def hasNext: Boolean = !these.isEmpty
    def next: A = 
      if (hasNext) {
        val result = these.head; these = these.tail; result
      } else Iterator.empty.next
    def fork = new ForkableIterator(these)
}

使用：

scala> val it = new ForkableIterator(List(1,2,3,4,5,6))
it: ForkableIterator[Int] = non-empty iterator

scala> it.next
res72: Int = 1

scala> val it2 = it.fork
it2: ForkableIterator[Int] = non-empty iterator

scala> it2.next
res73: Int = 2

scala> it2.next
res74: Int = 3

scala> it.next
res75: Int = 2

我看过为 HashMap 执行此操作，但它似乎更复杂（部分原因是根据集合大小有不同的映射实现） 。因此，最好在 yourMap.toList 上使用上述实现。

Answer 3

正如雷克斯所说，不可能在不破坏迭代器的情况下复制它。也就是说，duplicate 有什么问题吗？

var list = List(1,2,3,4,5)
var it1 = list.iterator
it1.next()

val (it1a, it1b) = it1.duplicate
it1 = it1a
var it2 = it1b
it2.next()

println(it1.next())

Answer 4

我认为这是一个很好的问题，遗憾的是很多人不理解这个问题的价值。在大数据时代，很多情况下我们有一个流，而不是无法收集或装入内存的已分配数据列表。从一开始就重复它的成本也很高。如果我们需要对数据进行两次（或多次）单独计算，我们该怎么办？例如，我们可能需要使用已经编写的函数来计算最小值、最大值、总和、md5 等，并且在不同线程中仅传递一次。

一般的解决方案是使用 Akka-Stream。这样就可以了。
但是 Iterator 是否可能，这是 Java/Scala 中表示此类流数据源的最简单方法？
答案是肯定的，尽管我们“无法完全独立地进行原始和复制”，这意味着我们必须同步每个消费者线程的消耗速度。 （Akka-Stream 使用背压和一些中间缓冲区来实现这一点）。

所以这是我的简单解决方案：使用 Phaser。有了它，我们可以对一次性源进行迭代器包装。该对象将在每个消费者线程中用作简单的迭代器。使用它你可以提前知道消耗线程的数量。此外，每个消费者线程必须耗尽源直到最后，以避免所有的挂起（例如使用flush()方法）。

import java.util.concurrent.Phaser
import java.util.concurrent.atomic.AtomicBoolean

// it0 - input source iterator
// num - exact number of consuming threads. We have to know it in advance.
case class ForkableIterator[+A]( it0: Iterator[A], num: Int ) extends Phaser(num) with Iterator[A] {

  val it = it0.flatMap( Stream.fill(num)(_) )  // serial replicator

  private var hasNext0 = new AtomicBoolean( it0.hasNext )
  override def hasNext: Boolean = hasNext0.get()

  override def next(): A = {
    arriveAndAwaitAdvance()
    val next = it.synchronized {
      val next = it.next()
      if (hasNext0.get) hasNext0.set(it.hasNext)
      next
    }
    arriveAndAwaitAdvance() // otherwise the tasks locks at the end the last data element
    next
  }

  // In case that a consumer gives up to read before the end of its source data stream occurs
  // it HAVE to drain the last to avoid block others. (Note: Phaser has no "unregister" method?).
  // Calling it may be avoided if all consumers read exactly the same amount of data,
  // e.g. until the very end of it.
  def flush(): Unit = while (hasNext) next()
}

PS 这个“ForkableIterator”被我成功地与 Spark 一起使用，对长源数据流执行多个独立的聚合。在这种情况下，我不需要手动创建线程。您还可以使用 Scala Futures / Monix Tasks 等。

PSPS 我现在重新检查 JDK Phaser 规范，发现它实际上有一个名为reachAndDeregister()的“取消注册”方法。因此，如果消费者完成，请使用它而不是flush()。