Akka Stream 写入多个文件

Question

如 Documentation 中所述，可以将可重用组件（例如 lineSink 写入文件）。

我想知道如何重写文档中的 lineSink 组件以实现以下场景：

val source = Source(1 to 10)
val flow = Flow[Int].map(_.toString)

/** Rewrite of the lineSink to enable new files with content */
val fileSink = Sink[String, Future[IOResult]] = 
  Flow[String]
    .map(s => ByteString(s + "\n"))
    .toMat(FileIO.toPath(Paths.get(filename)))(Keep.right)

/** After running the graph i want to have a new file for each item processed */
val graph = source.via(flow).to(sink)

Answer 1

您将需要为每个新的传入元素具体化一个新流。随着 FileIO.toPath 接收器具体化为 Future，您可以使用 mapAsync（具有合理的并行性选择）来实现此目的。请参见下面的示例：

  val source = Source(1 to 10)
  val flow: Flow[Int, String, NotUsed] = Flow[Int].map(_.toString)

  val fileFlow: Flow[String, IOResult, NotUsed] =
    Flow[String].mapAsync(parallelism = 4){ s ⇒
      Source.single(ByteString(s)).runWith(FileIO.toPath(Paths.get(fileName)))
    }

  val fileSink: Sink[IOResult, Future[Done]] = Sink.foreach[IOResult]{println}

  val graph = source.via(flow).via(fileFlow).to(fileSink)

请注意，您需要为每个传入元素生成适当的 fileName。你需要想出一个方法来做到这一点。

Akka Stream 写入多个文件

Akka Stream write to multiple Files

akka

akka-stream