scala - 如何动态生成大数据流

Question

我必须即时生成一个大文件。读取数据库并将其发送给客户端。
我阅读了一些文档，我做到了

val streamContent: Enumerator[Array[Byte]] = Enumerator.outputStream {
        os => 
              // new PrintWriter() read from database and for each record 
              // do some logic and write
              // to outputstream
      }
      Ok.stream(streamContent.andThen(Enumerator.eof)).withHeaders(
              CONTENT_DISPOSITION -> s"attachment; filename=someName.csv"
        )

我对 Scala 的新手一般只有一周，所以不要以我的声誉为指导。

我的问题是:

1)这是最好的方法吗？我发现如果我有一个大文件，这将加载到内存中，并且在这种情况下也不知道块大小是多少，如果它会为每个 write() 发送就是不方便。

2) 我找到了这个方法 Enumerator.fromStream(data : InputStream, chunkedSize : int)好一点，因为它有一个块大小，但我没有 inputStream 原因我即时创建文件。

Answer 1

docs for Enumerator.outputStream 中有一条说明:

Not [sic!] that calls to write will not block, so if the iteratee that is being fed to is slow to consume the input, the OutputStream will not push back. This means it should not be used with large streams since there is a risk of running out of memory.

这是否会发生取决于您的情况。如果您可以并且将在几秒钟内生成千兆字节，您可能应该尝试不同的东西。我不确定是什么，但我会从 Enumerator.generateM() 开始.但是，在许多情况下，您的方法非常好。看看 at this example by Gaëtan Renaudeau for serving a Zip file that's generated on the fly in the same way you're using it :

val enumerator = Enumerator.outputStream { os =>
  val zip = new ZipOutputStream(os);
  Range(0, 100).map { i =>
    zip.putNextEntry(new ZipEntry("test-zip/README-"+i+".txt"))
    zip.write("Here are 100000 random numbers:\n".map(_.toByte).toArray)
    // Let's do 100 writes of 1'000 numbers
    Range(0, 100).map { j =>
      zip.write((Range(0, 1000).map(_=>r.nextLong).map(_.toString).mkString("\n")).map(_.toByte).toArray);
    }
    zip.closeEntry()
  }
  zip.close()
}
Ok.stream(enumerator >>> Enumerator.eof).withHeaders(
  "Content-Type"->"application/zip", 
  "Content-Disposition"->"attachment; filename=test.zip"
)

请记住 Ok.stream已被 Ok.chunked 取代在较新版本的 Play 中，以防您想升级。

至于块大小，你总是可以使用 Enumeratee.grouped 收集一堆值并将它们作为一个块发送。

val grouper = Enumeratee.grouped(  
  Traversable.take[Array[Double]](100) &>> Iteratee.consume()  
)

然后你会做类似的事情

Ok.stream(enumerator &> grouper >>> Enumerator.eof)