Scalaz 7 Iteratee 处理大型 zip 文件(OutOfMemoryError)

Question

我正在尝试使用 scalaz iteratee 包来处理恒定空间中的大型 zip 文件。我有一个长时间运行的过程，我需要对 zip 文件中的每个文件执行。这些进程可以(并且应该)并行运行。

我创建了一个 EnumeratorT每个充气ZipEntry成File目的。签名看起来像:

def enumZipFile(f:File):EnumeratorT[IoExceptionOr[IO[File]], IO]

我想附上一个 IterateeT这将对每个文件执行长时间运行的过程。我基本上最终会得到类似的东西:

type IOE[A] = IoExceptionOr[A]

def action(f:File):IO[List[Promise[IOE[File]]]] = (
  consume[Promise[IOE[File]], IO, List] %=
  map[IOE[File], Promise[IOE[File]], IO](longRunningProcess) %=
  map[IOE[IO[File]], IOE[File], IO](_.unsafePerformIO) &=
  enumZipFile(f)
).run

def longRunningProcess:(iof:IOE[File]):Promise[IOE[File]] =
  Promise { Thread.sleep(5000); iof }

当我尝试运行它时:

action(new File("/really/big/file.zip")).unsafePerformIO.sequence.get

我得到一个 java.lang.OutOfMemoryError: Java heap space信息。这对我来说很有意义，因为它试图建立一个庞大的列表来记录所有这些 IO和 Promise对象。

几个问题:

有没有人对如何避免这种情况有任何想法？感觉我是在错误地处理问题，因为我真的只关心 longRunningProcess因为它的副作用。

是Enumerator方法在这里是错误的方法？

我几乎没有想法，所以任何事情都会有所帮助。

谢谢!

更新 #1

这是堆栈跟踪:

[error] java.lang.OutOfMemoryError: Java heap space
[error]         at scalaz.Free.flatMap(Free.scala:46)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)

我目前正在接受 nadavwr 的建议，以确保一切都像我想的那样。我会报告任何更新。

更新 #2

使用以下两个答案的想法，我找到了一个不错的解决方案。正如 huynhjl 所建议的(我使用 nadavwr 分析堆转储的建议进行了验证)， consume导致每次膨胀 ZipEntry保存在内存中，这就是进程内存不足的原因。我改了 consume至 foldM并更新了长时间运行的进程以仅返回 Promise[IOE[Unit]]而不是对文件的引用。这样我最后就有了所有 IoExceptions 的集合。这是工作解决方案:

def action(f:File):IO[List[Promise[IOE[Unit]]]] = (
  foldM[Promise[IOE[Unit]], IO, List[Promise[IOE[Unit]]]](List.empty)((acc,x) => IO(x :: acc)) %=
  map[IOE[File], Promise[IOE[Unit]], IO](longRunningProcess) %=
  map[IOE[IO[File]], IOE[File], IO](_.unsafePerformIO) &=
  enumZipFile(f)
).run

def longRunningProcess:(iof:IOE[File]):Promise[IOE[Unit]] =
  Promise { Thread.sleep(5000); iof.map(println) }

该解决方案在异步上传每个条目的同时膨胀了它们。最后，我有一个巨大的完成列表 Promise包含任何错误的对象。我仍然不完全相信这是 Iteratee 的正确用法，但我现在确实有几个可重用的、可组合的部分，我可以在我们系统的其他部分中使用它们(这对我们来说是一种非常常见的模式)。

感谢你的帮助!

Answer 1

不要使用 consume .查看我最近的另一个回答:How to use IO with Scalaz7 Iteratees without overflowing the stack?
foldM可能是更好的选择。

还尝试将文件映射到其他内容(如成功返回代码)，以查看是否允许 JVM 垃圾收集膨胀的 zip 条目。