scala - 并行 Scala 流的内存消耗

Question

我编写了一个 Scala (2.9.1-1) 应用程序，它需要处理数据库查询中的几百万行。我正在转换 ResultSet到 Stream使用对我的 previous questions 之一的回答中显示的技术:

class Record(...)

val resultSet = statement.executeQuery(...)

new Iterator[Record] {
  def hasNext = resultSet.next()
  def next = new Record(resultSet.getString(1), resultSet.getInt(2), ...)
}.toStream.foreach { record => ... }

这非常有效。

自体 foreach闭包非常占用 CPU，作为对函数式编程实用性的证明，如果我添加一个 .par之前 foreach ，除了确保闭包的主体是线程安全的(它以函数风格编写，除了打印到线程安全日志之外，没有可变数据)之外，无需其他任何努力即可并行运行闭包。

但是，我担心内存消耗。是 .par导致整个结果集加载到 RAM 中，还是并行操作只加载与事件线程一样多的行？我已将 4G 分配给 JVM(64 位，带有 -Xmx4g)，但将来我将在更多行上运行它，并担心最终会出现内存不足的情况。

是否有更好的模式以功能方式进行这种并行处理？我一直在向我的同事展示这个应用程序，作为函数式编程和多核机器值(value)的一个例子。

Answer 1

如果你看 scaladoc of Stream ，你会注意到 par 的定义类是 Parallelizable特质...而且，如果您查看 source code of this trait ，您会注意到它从原始集合中取出每个元素并将它们放入组合器中，因此，您将每一行加载到 ParSeq 中。 :

  def par: ParRepr = {
    val cb = parCombiner
    for (x <- seq) cb += x
    cb.result
  }

  /** The default `par` implementation uses the combiner provided by this method
   *  to create a new parallel collection.
   *
   *  @return  a combiner for the parallel collection of type `ParRepr`
   */
  protected[this] def parCombiner: Combiner[A, ParRepr]

一个可能的解决方案是显式并行化您的计算，例如多亏了 actor。你可以看看 this example例如，从 akka 文档中，这可能对您的上下文有所帮助。