将Traversable[T]转换为没有遍历或堆栈溢出的Stream[T]

Question

我正在使用一个库，它提供了一个通过数据库结果进行页面处理的TraversableT。我希望避免将整个事件加载到内存中，因此我尝试将其转换为StreamT。

据我所知，内置的"asStream“方法将整个可遍历的代码加载到缓冲区中，这违背了我的目的。我的尝试(下面)击中了一个大的结果的StackOverflowException，我不知道为什么。有人能帮我了解一下发生了什么吗？谢谢!

def asStream[T](traversable: => Traversable[T]): Stream[T] = {
  if (traversable.isEmpty) Empty
  else {
    lazy val head = traversable.head
    lazy val tail = asStream(traversable.tail)
    head #:: tail
  }
}

下面是一个完整的示例，它根据@SCouto的建议复制这个

import scala.collection.immutable.Stream.Empty

object StreamTest {
  def main(args: Array[String]) = {
    val bigVector = Vector.fill(90000)(1)
    val optionStream = asStream(bigVector).map(v => Some(v))
    val zipped = optionStream.zipAll(optionStream.tail, None, None)
  }

  def asStream[T](traversable: => Traversable[T]): Stream[T] = {
    @annotation.tailrec
    def loop(processed: => Stream[T], pending: => Traversable[T]): Stream[T] = {
      if (pending.isEmpty) processed
      else {
        lazy val head = pending.head
        lazy val tail = pending.tail
        loop(processed :+ head, tail)
      }
    }

    loop(Empty, traversable)
  }
}

编辑:在@SCouto的一些有趣的想法之后，我了解到这也可以用蹦床来完成，以将结果保持为一个按原来顺序排列的StreamT。

object StreamTest {
  def main(args: Array[String]) = {
    val bigVector = Range(1, 90000).toVector
    val optionStream = asStream(bigVector).map(v => Some(v))
    val zipped = optionStream.zipAll(optionStream.tail, None, None)
    zipped.take(10).foreach(println)
  }

  def asStream[T](traversable: => Traversable[T]): Stream[T] = {
    sealed trait Traversal[+R]
    case class More[+R](result: R, next: () => Traversal[R]) extends Traversal[R]
    case object Done extends Traversal[Nothing]

    def next(currentTraversable: Traversable[T]): Traversal[T] = {
      if (currentTraversable.isEmpty) Done
      else More(currentTraversable.head, () => next(currentTraversable.tail))
    }

    def trampoline[R](body: => Traversal[R]): Stream[R] = {
      def loop(thunk: () => Traversal[R]): Stream[R] = {
        thunk.apply match {
          case More(result, next) => Stream.cons(result, loop(next))
          case Done => Stream.empty
        }
      }
      loop(() => body)
    }

    trampoline(next(traversable))
  }
}

Answer 1

试试这个：

  def asStream[T](traversable: => Traversable[T]): Stream[T] = {

    @annotation.tailrec
    def loop(processed: Stream[T], pending: Traversable[T]): Stream[T] = {
      if (pending.isEmpty) processed
      else {
        lazy val head = pending.head
        lazy val tail = pending.tail
        loop(head #:: processed, tail)
      }
    }

    loop(Empty, traversable)
  }

要点是确保递归调用是递归函数的最后一个动作。

为了确保这一点，可以同时使用嵌套方法(在示例中称为loop )和tailrec注释，以确保方法是尾安全的。

您可以在这个令人敬畏的答案( 这里 )中找到有关尾部rec 这里的信息

编辑问题是，我们在流的末尾添加了元素。如果您将它作为Stream的头添加，就像在您的示例中一样，它将很好地工作。我更新了密码。请进行测试，并让我们知道结果。

我的测试：

scala> val optionStream = asStream(Vector.fill(90000)(1)).map(v => Some(v))
optionStream: scala.collection.immutable.Stream[Some[Int]] = Stream(Some(1), ?)

scala> val zipped = optionStream.zipAll(optionStream.tail, None, None)
zipped: scala.collection.immutable.Stream[(Option[Int], Option[Int])] = Stream((Some(1),Some(1)), ?)

EDIT2:

根据您的评论，并考虑您所说的fpinscala的例子。我觉得这可能对你有帮助。重点是创建一个带有延迟计算的案例类结构。其中头部是一个单一的元素，而尾巴是可遍历的。

sealed trait myStream[+T] {
  def head: Option[T] = this match {
    case MyEmpty => None
    case MyCons(h, _) => Some(h())
  }


  def tail: myStream[T] = this match {
      case MyEmpty => MyEmpty
      case MyCons(_, t) => myStream.cons(t().head, t().tail)
    }
}
case object MyEmpty extends myStream[Nothing]
case class MyCons[+T](h: () => T, t: () => Traversable[T]) extends myStream[T]


object myStream {

  def cons[T](hd: => T, tl: => Traversable[T]): myStream[T] = {
    lazy val head = hd
    lazy val tail = tl

    MyCons(() => head, () => tail)
  }

  def empty[T]: myStream[T] = MyEmpty

  def apply[T](as: T*): myStream[T] = {
    if (as.isEmpty) empty
    else cons(as.head, as.tail)
  }
}

一些快速测试：

  val bigVector = Vector.fill(90000)(1)
myStream.cons(bigVector.head, bigVector.tail)
res2: myStream[Int] = MyCons(<function0>,<function0>)

回收头：

res2.head
res3: Option[Int] = Some(1)

而尾巴：

res2.tail
res4: myStream[Int] = MyCons(<function0>,<function0>)

EDIT3

执行部分的蹦床解决方案：

 def asStream[T](traversable: => Traversable[T]): Stream[T] = {
    sealed trait Traversal[+R]
    case class More[+R](result: R, next: () => Traversal[R]) extends Traversal[R]
    case object Done extends Traversal[Nothing]

    def next(currentTraversable: Traversable[T]): Traversal[T] = {
      if (currentTraversable.isEmpty) Done
      else More(currentTraversable.head, () => next(currentTraversable.tail))
    }

    def trampoline[R](body: => Traversal[R]): Stream[R] = {
      def loop(thunk: () => Traversal[R]): Stream[R] = {
        thunk.apply match {
          case More(result, next) => Stream.cons(result, loop(next))
          case Done => Stream.empty
        }
      }
      loop(() => body)
    }

    trampoline(next(traversable))
      }
    }