Scala：如何“预加载”惰性迭代器的内容？

Question

假设我有一个惰性迭代器[Item]。仅当我们迭代迭代器时，才会延迟创建 items 对象。这些物品的制作成本很高。

我想将此迭代器序列化为 JSON 数组。它可以工作（使用 Jackson scala 模块），但在我看来效率不够。

据我了解，目前的工作方式如下：

• 计算下一项
• 序列化项目
• 计算下一项
• 序列化项目
• 计算下一项
• 序列化项目

我希望项目的计算和项目的序列化并行进行。

我想要一个迭代器，它会在读取下一项时开始计算定义数量的下一项。

例如，我希望在执行 iterator.next() 时，在幕后计算接下来的 50 个项目，而不会阻塞迭代线程（它应该只等待下一个可用元素）。

我见过“BufferedIterator”，但它并不是我所需要的，因为我真的不想明确地查询“head”，而且我需要超过 1 个项目来预加载

知道如何实现这一点吗？

我也可以用 Stream 替换 Iterator 的解决方案，但由于内存使用率较低，我更喜欢 Iterator

Answer 1

如果我对您的问题的理解正确，这里有一个示例说明您可以做什么。您可以将每个项目计算包装在Future 中，这样您就可以在不阻塞的情况下迭代输入流，并在每个块准备好后对其进行处理/序列化。我将在 REPL 中执行此操作，并在评估每件作品时打印出来，这样您就可以看到每件事情发生的时间：

@ import concurrent._, ExecutionContext.Implicits.global
import concurrent._, ExecutionContext.Implicits.global

@ def futureItem(i: Int): Future[Int] = Future { 
  Thread.sleep(1000)
  println(s"item: ${i}")
  i 
}
defined function futureItem

@ val inputIterator = (1 to 9).toIterator.map(futureItem)
inputIterator: Iterator[Future[Int]] = non-empty iterator

因此计算每个项目至少需要 1 秒。现在我们要分块处理项目，这也需要一些时间：

@ def computeItemsChunk(items: Seq[Int]): Int = { 
  Thread.sleep(1000)
  val s = items.sum
  println(s"chunk ${items}: ${s}")
  s 
}
defined function computeItemsChunk

现在我们对输入流进行分组，应用Future.sequence 并计算块：

@ case object foo {
  val chunksIterator = inputIterator.grouped(3).map { futureItems => 
    Future.sequence(futureItems).map(computeItemsChunk) 
  }
}
defined object foo

（我在一个对象中定义它，因为否则，分组（或其他东西）将强制评估第一个块）。现在让我们看看它是如何评估的：

@ Await.result(Future.sequence(foo.chunksIterator), Duration.Inf)
item: 2
item: 3
item: 4
item: 1
item: 7
item: 6
chunk List(1, 2, 3): 6
item: 5
item: 8
chunk List(4, 5, 6): 15
item: 9
chunk List(7, 8, 9): 24
res5: Iterator[Int] = non-empty iterator 

您可以看到，一旦项目可用并且迭代器前进而不等待每个块评估，就会计算块。