支持流处理的OutOfMemoryException

Question

我编写了一个函数，在给定Stream[Long]的情况下，它将过滤出大于4,000,000的数字。

  def filterLt4Mil(xs: Stream[Long]) = {
    @tailrec
    def go(xs: Stream[Long], acc: Stream[Long]): Stream[Long] = xs match {
      case Stream() => acc
      case a #:: as if(a < 4000000L) => go(as, acc :+ a)
      case a #:: as if(a > 4000000L) => go(as, acc)
    }
    go(xs, Stream[Long]())
  }

但是，当我传入一个范围从0到10,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000,000的OutOfMemoryException。

scala> val x = Stream.range(0,10000000L)
x: scala.collection.immutable.Stream[Long] = Stream(0, ?)

scala> filterLt4Mil(x)
java.lang.OutOfMemoryError: GC overhead limit exceeded
        at scala.collection.immutable.List.toStream(List.scala:312)

因为filterLt4Mil使用了尾部调用优化，所以我的理解是堆栈不应该溢出。

但是，为什么会出现这种OutOfMemoryException？我怎样才能防止它呢？

Answer 1

您的函数有一些问题。

您的go函数具有颠倒输出流中元素的效果，因为您将其追加到累加器的末尾。

此外，附加而不是预先附加的动作会导致整个累加器被物化到一个列表中，然后在迭代中的每个步骤中复制该列表。这就是导致OutOfMemoryError的原因，因为您的实现需要二次内存。

此外，由于您没有处理a == 4000000L的情况，所以您的match代码块并不是详尽的。

如果您想要做的就是过滤出大于4000000的元素，这就足够了：

xs filterNot {_ > 4000000L}

Answer 2

好吧，已经有一个用于流的filter方法，但我猜您想自己来做。

Streams有严格的头部和懒惰的尾巴。调用Stream.range(0,10000000L)将只有0在内存中，因为其余的将延迟求值。

但是在您的filterLt4Mil方法中，您将迭代整个流，导致该流中的所有元素都被求值，从而导致从0到10000000L的所有数字都被存储在内存中……存储这样的数量可能会导致OutOfMemoryException。

例如：x.forall(_ >= 0)也会导致OutOfMemoryException，forall还必须在满足_ >= 0的同时遍历整个流。

您可能希望延迟应用您的filter方法。请看一下源代码。可能会让你对它的工作原理有所了解：(http://harrah.github.io/browse/samples/library/scala/collection/immutable/Stream.scala.html)

override final def filter(p: A => Boolean): Stream[A] = {
  // optimization: drop leading prefix of elems for which f returns false
  var rest = this dropWhile (!p(_))
  if (rest.isEmpty) Stream.Empty
  else new Stream.Cons(rest.head, rest.tail filter p)
}

关于你的代码还有一个问题。因为它必须遍历整个流，所以您可以猜测无限流会发生什么。

Answer 3

我认为你没有搞砸堆栈--看起来你的堆快用完了(为了确认，切换到使用Int(s)而不是Long(s))。scala编译器应该知道如何使用tail-recurse优化这个函数。

10m的长度是64m。如果你正在创建一个64m列表中的10m个实例，那么你就有麻烦了。

此外，您可能希望使用惰性流，而不仅仅是尾部递归，以便对此函数进行内存优化。64m本身并不是很难，所以这个优化可能不是必需的。

在我看来，您要么创建了XS参数的10m个副本，要么创建了ACC参数的10M个副本。

在我看来，xs:参数似乎很懒；但是您在函数外部保留了对它的引用，而streams在scala中是记忆器。

所以我敢打赌，这个参数就是你会得到数百万个副本的那个(即，scala编译器不知道要去掉的中间值)。另一个参数应该在每次调用时进行垃圾回收(或者，因为我们讨论的是尾递归，所以在内存列表中优化为1，64m长)