Scala中无限流的嵌套迭代

Question

有时我发现自己想在 Scala for 推导中对无限流执行嵌套迭代，但指定循环终止条件可能有点棘手。有没有更好的方法来做这种事情？

我想到的用例是我不一定预先知道我正在迭代的每个无限流中需要多少元素（但显然我知道它不会是无限数）。假设每个流的终止条件可能以某种复杂的方式依赖于 for 表达式中其他元素的值。

最初的想法是尝试将流终止条件写为 for 表达式中的 if 过滤子句，但是在循环嵌套无限流时会遇到问题，因为没有办法使第一个无限流上的迭代短路，最终导致 OutOfMemoryError。我理解为什么会这样，考虑到 for 表达式如何映射到 ma​​p、flatMap 和 withFilter 方法调用 -我的问题是做这种事情是否有更好的习语（也许根本不涉及 for 理解）。

为了给出一个有点人为的例子来说明刚刚描述的问题，请考虑以下（非常幼稚的）代码来生成数字 1 和 2 的所有配对：

val pairs = for {
  i <- Stream.from(1) 
  if i < 3 
  j <- Stream.from(1) 
  if j < 3
} 
yield (i, j)

pairs.take(2).toList 
// result: List[(Int, Int)] = List((1,1), (1,2)) 

pairs.take(4).toList
// 'hoped for' result: List[(Int, Int)] = List((1,1), (1,2), (2,1), (2,2))
// actual result:
//  java.lang.OutOfMemoryError: Java heap space
//      at scala.collection.immutable.Stream$.from(Stream.scala:1105)

显然，在这个简单的示例中，可以通过将 if 过滤器移动到原始流上的 takeWhile 方法调用中轻松避免问题，如下所示：

val pairs = for {
  i <- Stream.from(1).takeWhile(_ < 3) 
  j <- Stream.from(1).takeWhile(_ < 3) 
}    
yield (i, j)

但出于问题的目的，请想象一个更复杂的用例，其中流终止条件不能轻易移动到流表达式本身。

Answer 1

一种可能性是将Stream 包装到您自己的类中，以不同方式处理filter，在本例中为takeWhile：

import scala.collection._
import scala.collection.generic._

class MyStream[+A]( val underlying: Stream[A] ) {
  def flatMap[B, That](f: (A) => GenTraversableOnce[B])(implicit bf: CanBuildFrom[Stream[A], B, That]): That = underlying.flatMap(f);

  def map[B, That](f: (A) ⇒ B)(implicit bf: CanBuildFrom[Stream[A], B, That]): That = underlying.map(f);

  def filter(p: A => Boolean): Stream[A] = underlying.takeWhile(p);
  //                                       ^^^^^^^^^^^^^^^^^^^^^^^^
}

object MyStream extends App {
  val pairs = for {
    i <- new MyStream(Stream.from(1))
    if i < 3
    j <- new MyStream(Stream.from(1))
    if j < 3
  } yield (i, j);

  print(pairs.toList);
}

这打印出List((1,1), (1,2), (2,1), (2,2))。

Answer 2

我已经采纳了 Petr 的建议以提出我认为更普遍可用的解决方案，因为它不会限制 if 过滤器在 for 理解中的位置（尽管它有更多的语法开销）。

这个想法再次将底层流封装在一个包装器对象中，它代表 flatMap、map 和 filter 方法而不进行修改，但首先将 takeWhile 调用应用于底层流，并带有!isTruncated 的谓词，其中isTruncated 是属于包装器对象的字段。在任何时候对包装对象调用truncate 都会翻转isTruncated 标志并有效地终止对流的进一步迭代。这在很大程度上依赖于对底层流的 takeWhile 调用是延迟评估的事实，因此在迭代后期执行的代码可能会影响其行为。

缺点是您必须保留对希望能够在迭代中截断的流的引用，方法是将 || s.truncate 附加到过滤器表达式（其中s 是对包装流的引用）。您还需要确保在流中的每次新迭代之前在包装器对象上调用 reset（或使用新的包装器对象），除非您知道重复迭代每次的行为都相同。

import scala.collection._
import scala.collection.generic._

class TruncatableStream[A]( private val underlying: Stream[A]) {
  private var isTruncated = false;

  private var active = underlying.takeWhile(a => !isTruncated)

  def flatMap[B, That](f: (A) => GenTraversableOnce[B])(implicit bf: CanBuildFrom[Stream[A], B, That]): That = active.flatMap(f);

  def map[B, That](f: (A) => B)(implicit bf: CanBuildFrom[Stream[A], B, That]): That = active.map(f);

  def filter(p: A => Boolean): Stream[A] = active.filter(p);

  def truncate() = {
    isTruncated = true
    false
  }

  def reset() = {
    isTruncated = false
    active = underlying.takeWhile(a => !isTruncated)
  }
}

val s1 = new TruncatableStream(Stream.from(1))
val s2 = new TruncatableStream(Stream.from(1))

val pairs = for {
  i <- s1

  // reset the nested iteration at the start of each outer iteration loop 
  // (not strictly required here as the repeat iterations are all identical)
  // alternatively, could just write: s2 = new TruncatableStream(Stream.from(1))  
  _ = _s2.reset()      

  j <- s2
  if i < 3 || s1.truncate
  if j < 3 || s2.truncate
} 
yield (i, j)

pairs.take(2).toList  // res1: List[(Int, Int)] = List((1,1), (1,2))
pairs.take(4).toList  // res2: List[(Int, Int)] = List((1,1), (1,2), (2,1), (2,2))

毫无疑问，这可以改进，但它似乎是解决问题的合理方法。