如何将 Scala 流的内容写入文件？

Question

我有一个想要写入文件的 Scala 字节流。流中的数据过多，无法缓冲所有内存。

作为第一次尝试，我创建了一个类似于此的InputStream：

class MyInputStream(data: Stream[Byte]) extends InputStream {
  private val iterator = data.iterator
  override def read(): Int = if (iterator.hasNext) iterator.next else -1
}

然后我使用 Apache Commons 编写文件：

val source = new MyInputStream(dataStream)
val target = new FileOutputStream(file)
try {
  IOUtils.copy(source, target)
} finally {
  target.close
}

这可行，但我对性能不太满意。我猜想为每个字节调用MyInputStream.read 会带来很多开销。有没有更好的办法？

Answer 1

您可能（也可能不会！）误认为读取端是性能问题的根源。这可能是因为您使用的是无缓冲的 FileOutputStream(...)，从而强制对每个写入的字节进行单独的系统调用。

这是我的看法，快速而简单：

def writeBytes( data : Stream[Byte], file : File ) = {
  val target = new BufferedOutputStream( new FileOutputStream(file) );
  try data.foreach( target.write(_) ) finally target.close;
}

Answer 2

我推荐java.nio.file 包。使用Files.write，您可以将Bytes 的Arrays 写入从文件名构造的Path。

如何提供Bytes 由您决定。您可以使用.toArray 将Stream 转换为Array，或者您可以一次将一个（或几个）流中的take 字节转换为数组。

这是一个演示.toArray 方法的简单代码块。

import java.nio.file.{Files, Paths}

val filename: String = "output.bin"
val bytes: Stream[Byte] = ...
Files.write(Paths.get(filename), bytes.toArray)

Answer 3

您应该在 InputStream 实现中实现批量读取覆盖：

override def read(b: Array[Byte], off: Int, len: Int)

IOUtils.copy 使用该签名读取/写入 4K 块。

Answer 4

鉴于StreamIterator 一次读取一个字节可能是瓶颈，我设计了一种将流写入OutputStream 的方法，该方法不依赖于它并且希望更高效：

object StreamCopier {
  def copy(data: Stream[Byte], output: OutputStream) = {
    def write(d: Stream[Byte]): Unit = if (d.nonEmpty) {
      val (head, tail) = d.splitAt(4 * 1024)
      val bytes = head.toArray
      output.write(bytes, 0, bytes.length)
      write(tail)
    }
    write(data)
  }
}

编辑：通过在尾递归 write 函数中将 data 替换为 d 修复了一个错误。

此方法通过splitAt 使用递归方法将流拆分为第一个~4K 和其余部分，将其头部写入OutputStream 并在流的尾部递归，直到splitAt 返回一个空流。

既然您已经制定了性能基准，我会留给您判断这是否更有效。