小编ran*_*tic的帖子

我有一个文件列表.我想要:

1. 从所有这些作为单一来源阅读.
2. 文件应按顺序依次读取.(没有循环)
3. 在任何时候都不应该要求任何文件完全在内存中.
4. 从文件读取错误应该会折叠流.

感觉这应该有效:(Scala,akka-streams v2.4.7)

val sources = Seq("file1", "file2").map(new File(_)).map(f => FileIO.fromPath(f.toPath)
    .via(Framing.delimiter(ByteString(System.lineSeparator), 10000, allowTruncation = true))
    .map(bs => bs.utf8String)
  )
val source = sources.reduce( (a, b) => Source.combine(a, b)(MergePreferred(_)) )
source.map(_ => 1).runWith(Sink.reduce[Int](_ + _)) // counting lines

但是这会导致编译错误,因为它FileIO具有与之关联的物化值,并且Source.combine不支持它.

映射物化值让我想知道如何处理文件读取错误,但是编译:

val sources = Seq("file1", "file2").map(new File(_)).map(f => FileIO.fromPath(f.toPath)
    .via(Framing.delimiter(ByteString(System.lineSeparator), 10000, allowTruncation = true))
    .map(bs => bs.utf8String)
    .mapMaterializedValue(f => NotUsed.getInstance())
  )
val source = sources.reduce( (a, b) => Source.combine(a, b)(MergePreferred(_)) )
source.map(_ => …