你如何处理 Akka Flow 中的期货？

Question

我构建了一个定义流的 akka 图。我的 objective 是重新格式化我以后的回复并将其保存到文件中。流程如下所示：

val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
      import GraphDSL.Implicits._
      val balancer = builder.add(Balance[(HttpRequest, String)](6, waitForAllDownstreams = false))
      val merger = builder.add(Merge[Future[Map[String, String]]](6))
      val fileSink = FileIO.toPath(outputPath, options)
      val ignoreSink = Sink.ignore
      val in = Source(seeds)
      in ~> balancer.in
      for (i <- Range(0,6)) {
        balancer.out(i) ~>
          wikiFlow.async ~>
          // This maps to a Future[Map[String, String]]
          Flow[(Try[HttpResponse], String)].map(parseHtml) ~>
          merger
      }

      merger.out ~>
      // When we merge we need to map our Map to a file
      Flow[Future[Map[String, String]]].map((d) => {
        // What is the proper way of serializing future map
        // so I can work with it like a normal stream into fileSink?

        // I could manually do ->
        // d.foreach(someWriteToFileProcess(_))
        // with ignoreSink, but this defeats the nice
        // akka flow
      }) ~>
      fileSink

      ClosedShape
    })

我可以破解此工作流程，通过 foreach 将我未来的地图写入文件，但我担心这可能会以某种方式导致 FileIO 的并发问题，而且感觉不对。使用我们的 akka 流程处理期货的正确方法是什么？

Answer 1

创建涉及异步计算的 Flow 的最简单方法是使用 mapAsync.

所以...假设您要创建一个 Flow，它使用并行度为 5 的异步计算 mapper: Int => Future[String] 使用 Int 并生成 String。

val mapper: Int => Future[String] = (i: Int) => Future(i.toString)

val yourFlow = Flow[Int].mapAsync[String](5)(mapper)

现在，您可以根据需要在图表中使用此流程。

示例用法是，

val graph = GraphDSL.create() { implicit builder =>
  import GraphDSL.Implicits._

  val intSource = Source(1 to 10)

  val printSink = Sink.foreach[String](s => println(s))

  val yourMapper: Int => Future[String] = (i: Int) => Future(i.toString)

  val yourFlow = Flow[Int].mapAsync[String](2)(yourMapper)

  intSource ~> yourFlow ~> printSink

  ClosedShape
}

你如何处理 Akka Flow 中的期货？

How do you deal with futures in Akka Flow?

scala

future

akka

akka-stream