akka-stream - 如何使用 akka-streams 实现分页

Question

我需要逐行处理大文件并在每个项目上做一些繁重的工作（在 4 核 cpu 上），我认为代码正确：

implicit val system = ActorSystem("TestSystem")
implicit val materializer = ActorMaterializer()
import system.dispatcher

val sink = Sink.foreach[String](elem => println("element proceed"))

FileIO.fromPath(Paths.get("file.txt"))
  .via(Framing.delimiter(ByteString("\n"), 64).map(_.utf8String))
  .mapAsync(4)(v =>
    //long op
    Future {
      Thread.sleep(500)
      "updated_" + v
    })
  .to(sink)
  .run()

但我想要输出如下：

100 element proceed
200 element proceed
300 element proceed
357 element proceed. done

如何实施？

标签： akka-stream

Answer 1

您可以使用Flow.grouped：

val groupSize = 100

val groupedFlow = Flow[String].grouped(groupSize)

现在可以在您的之前或之后注入此流程mapAsync：

FileIO.fromPath(Paths.get("file.txt"))
      .via(Framing.delimiter(ByteString("\n"), 64).map(_.utf8String))
      .via(groupedFlow)
  ...