Akka Streams 文件处理和终止

Question

我有以下 sn-p，它读取一个 CSV 文件并将一些内容打印到控制台：

def readUsingAkkaStreams = {

    import java.io.File
    import akka.stream.scaladsl._
    import akka.actor.ActorSystem
    import akka.stream.ActorMaterializer
    import java.security.MessageDigest


    implicit val system = ActorSystem("Sys")
    implicit val materializer = ActorMaterializer()

    val file = new File("/path/to/csv/file.csv")

    val fileSource = FileIO.fromFile(file, 65536)

    val flow = fileSource.map(chunk => chunk.utf8String)

    flow.to(Sink.foreach(println(_))).run
  }

我现在对此有一些疑问：

1. 块大小是以字节为单位的大小。内部如何处理？我的意思是我最终会遇到一个块可能只包含一行中的部分元素的情况吗？

2. 此流如何终止？现在没有！我想让它知道它已经完全读取了文件并且应该触发停止信号！有没有一种机制可以做到这一点？

编辑 1：根据下面帖子的建议，我收到如屏幕截图所示的错误消息！

编辑 2：

通过设置 maximumFrameLength 以匹配最大块大小的大小（即 65536）来消除错误。

val file = new File("/path/to/csf/file.csv")
val chunkSize = 65536
val fileSource = FileIO.fromFile(file, chunkSize).via(Framing.delimiter(
  ByteString("\n"),
  maximumFrameLength = chunkSize,
  allowTruncation = true))

Answer 1

1.根据docs:

发出的元素是 chunkSize 大小的 ByteString 元素，最终元素除外，它的大小将最大为 chunkSize。

FileIO 源将换行视为任何其他字符。所以是的，您可能会在一个块中看到 CSV 行的第一部分，而在另一个块中看到第二部分。如果这不是您想要的，您可以使用Framing.delimiter 重组ByteString 流的分块方式（有关更多信息，请参阅docs）。

附带说明，FileIO.fromFile 已被弃用，最好使用FileIO.fromPath。

一个例子是：

val fileSource = FileIO.fromPath(...)
  .via(Framing.delimiter(
    ByteString("\n"),
    maximumFrameLength = 256,
    allowTruncation = true))

2. 接收器具体化为Future，当流终止时，您可以映射到做某事：

val result: Future[IOResult] = flow.runWith(Sink.foreach(println(_)))

result.onComplete(...)