这是一个使用 Scalaz 7 库的快速迭代示例,演示了您感兴趣的属性:常量内存和堆栈使用。
问题
首先假设我们有一个大文本文件,每行都有一串十进制数字,我们想要找到所有至少包含 20 个零的行。我们可以像这样生成一些样本数据:
val w = new java.io.PrintWriter("numbers.txt")
val r = new scala.util.Random(0)
(1 to 1000000).foreach(_ =>
w.println((1 to 100).map(_ => r.nextInt(10)).mkString)
)
w.close()
现在我们有了一个名为numbers.txt
. 让我们用以下命令打开它BufferedReader
:
val reader = new java.io.BufferedReader(new java.io.FileReader("numbers.txt"))
它不是太大(约 97 兆字节),但足以让我们很容易地看到我们的内存使用是否在我们处理它时实际上保持不变。
设置我们的枚举器
首先是一些进口:
import scalaz._, Scalaz._, effect.IO, iteratee.{ Iteratee => I }
还有一个枚举器(请注意,为了方便起见,我将IoExceptionOr
s更改为 s ):Option
val enum = I.enumReader(reader).map(_.toOption)
Scalaz 7 当前不提供枚举文件行的好方法,因此我们一次将文件分块处理一个字符。这当然会非常缓慢,但我不会在这里担心,因为这个演示的目标是展示我们可以在恒定内存中处理这个大文件,而不会破坏堆栈。该答案的最后一部分提供了一种性能更好的方法,但在这里我们将仅在换行符处进行拆分:
val split = I.splitOn[Option[Char], List, IO](_.cata(_ != '\n', false))
splitOn
而且,如果采用指定不拆分位置的谓词这一事实使您感到困惑,那么您并不孤单。split
是我们的第一个枚举示例。我们将继续将我们的枚举器包装在其中:
val lines = split.run(enum).map(_.sequence.map(_.mkString))
现在我们在monad中有一个Option[String]
s的枚举器。IO
使用枚举对象过滤文件
接下来是我们的谓词——请记住,我们说过我们想要至少有 20 个零的行:
val pred = (_: String).count(_ == '0') >= 20
我们可以把它变成一个过滤枚举器并将我们的枚举器包装在其中:
val filtered = I.filter[Option[String], IO](_.cata(pred, true)).run(lines)
我们将设置一个简单的操作,仅打印通过此过滤器的所有内容:
val printAction = (I.putStrTo[Option[String]](System.out) &= filtered).run
当然,我们还没有真正读过任何东西。为此,我们使用unsafePerformIO
:
printAction.unsafePerformIO()
现在我们可以看到Some("0946943140969200621607610...")
s 慢慢滚动,而我们的内存使用量保持不变。它很慢,错误处理和输出有点笨拙,但我认为对于大约九行代码来说还不错。
从迭代器获取输出
那是foreach
-ish 用法。我们还可以创建一个更像折叠的迭代器——例如收集通过过滤器的元素并将它们返回到一个列表中。只需重复上面的所有内容直到printAction
定义,然后改写:
val gatherAction = (I.consume[Option[String], IO, List] &= filtered).run
启动该操作:
val xs: Option[List[String]] = gatherAction.unsafePerformIO().sequence
现在去喝杯咖啡(可能需要离得很远)。当你回来时,你要么有一个None
(在IOException
某个地方的情况下),要么有一个Some
包含 1,943 个字符串的列表。
自动关闭文件的完整(更快)示例
为了回答您关于关闭阅读器的问题,这里有一个完整的工作示例,大致相当于上面的第二个程序,但有一个负责打开和关闭阅读器的枚举器。它也快得多,因为它读取的是行,而不是字符。首先是导入和几个辅助方法:
import java.io.{ BufferedReader, File, FileReader }
import scalaz._, Scalaz._, effect._, iteratee.{ Iteratee => I, _ }
def tryIO[A, B](action: IO[B]) = I.iterateeT[A, IO, Either[Throwable, B]](
action.catchLeft.map(
r => I.sdone(r, r.fold(_ => I.eofInput, _ => I.emptyInput))
)
)
def enumBuffered(r: => BufferedReader) =
new EnumeratorT[Either[Throwable, String], IO] {
lazy val reader = r
def apply[A] = (s: StepT[Either[Throwable, String], IO, A]) => s.mapCont(
k =>
tryIO(IO(reader.readLine())).flatMap {
case Right(null) => s.pointI
case Right(line) => k(I.elInput(Right(line))) >>== apply[A]
case e => k(I.elInput(e))
}
)
}
现在是枚举器:
def enumFile(f: File): EnumeratorT[Either[Throwable, String], IO] =
new EnumeratorT[Either[Throwable, String], IO] {
def apply[A] = (s: StepT[Either[Throwable, String], IO, A]) => s.mapCont(
k =>
tryIO(IO(new BufferedReader(new FileReader(f)))).flatMap {
case Right(reader) => I.iterateeT(
enumBuffered(reader).apply(s).value.ensuring(IO(reader.close()))
)
case Left(e) => k(I.elInput(Left(e)))
}
)
}
我们准备好了:
val action = (
I.consume[Either[Throwable, String], IO, List] %=
I.filter(_.fold(_ => true, _.count(_ == '0') >= 20)) &=
enumFile(new File("numbers.txt"))
).run
现在阅读器将在处理完成后关闭。