如何使用 Akka 流将 Flow1 物化值传递和消费到 Flow2 中

Question

我正在学习 Akka 并试图了解 GraphStage 的工作原理，但我在 ma​​terialized value 用法方面遇到了困难。

我想帮助解决以下问题：

如何将第一流的物化价值消耗到第二流中？

如有请提供指点

谢谢

Answer 1

我花了一些时间思考你的问题。简而言之，我的回答是：根据我对 akka-stream 的理解，您的问题毫无意义。下面是一个扩展的解释。另外我的知识有限，如果我假设有错误，请发表评论，让我们一起学习。

在考虑如何使用Flow 的物化值之前，我们首先需要有一个Flow 的示例，其物化值不同于NotUsed - 大多数开箱即用的流程中使用的示例。 所以，我编写了一个简单的Flow，它只传递所有上游元素，但嵌入了一个计数器，用于计算传递的总元素。 代码如下：

class FlowWithCounter extends GraphStageWithMaterializedValue[FlowShape[Int, Int], Future[Int]]{
    val in:Inlet[Int] = Inlet("Inlet_of_FlowWithCounter")
    val out:Outlet[Int] = Outlet("Outlet of FlowWithCounter")

    override def shape: FlowShape[Int, Int] = FlowShape(in, out)

    override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Int]) = {
      val materializedValue = Promise[Int]()
      val logic = new GraphStageLogic(shape) {
        private var counter: Int = 0

        setHandler(
          in,
          new InHandler {
            override def onPush(): Unit = {
              val elem = grab(in)
              counter += 1
              push(out, elem)
            }

            override def onUpstreamFinish(): Unit = {
              materializedValue.success(counter)
              super.onUpstreamFinish()
            }
          }
        )

        setHandler(out, new OutHandler {
          override def onPull(): Unit = {
            pull(in)
          }
        })
      }
      (logic, materializedValue.future)
    }
  }

如您所见，物化值类型是 Future[Int]。因此，当流完成时，如果我们指示物化器这样做，物化器将能够获取该值并使用它。

现在有几个重要的观察结果：

1. createLogicAndMaterializedValue 方法在我们运行流时不会被流 Materializer 调用。所以我们应该在那一刻返回一些指针，它将在流完成时刻保存物化值。这通常以Future[T] 的形式自然完成。

2. 如上所述here：

   // 所有状态必须在 GraphStageLogic 内，

   // 永远不会在封闭的 GraphStage 中。

   // 这个状态是安全的，可以从所有的访问和修改

   // GraphStageLogic 提供的回调和

   // 注册的处理程序。

   在我们的示例中，这意味着我们只能访问 new GraphStageLogic(shape) {...} 构造函数中的 counter 变量。我们不能在返回 (logic, materializedValue.future) 对的那一行引用 counter 本身。

我在 DSL 中没有找到任何方法来查询流组件的具体化值。只有借助toMat、viaMat 等方法，我们才能告诉物化器将组件的物化值传播到下游。并且只有借助run 这样的方法，我们才能最终提取整个流的最终物化值。

注意到我们的示例流程完成了带有 counter 值的承诺

override def onUpstreamFinish(): Unit = {
  materializedValue.success(counter)
  super.onUpstreamFinish()
}

当上游完成时。

以下是我们实际连接此示例流、通过流阶段传播其物化值并最终在整个流完成时使用它的方式：

import akka.stream.stage._

object Main {
  import scala.concurrent._
  import akka._
  import akka.actor._
  import akka.stream._
  import akka.stream.scaladsl._

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

  class FlowWithCounter extends GraphStageWithMaterializedValue[FlowShape[Int, Int], Future[Int]]{....}

  def main(args:Array[String]):Unit = {
    val source:Source[Int, NotUsed] = Source(15 to 25)
    val flow:Flow[Int, Int, Future[Int]] = Flow.fromGraph(new FlowWithCounter)
    val sink:Sink[Int, Future[Done]] = Sink.foreach(println)

    val completion:(Future[Int], Future[Done]) = ((source.viaMat(flow)(Keep.right)).toMat(sink)(Keep.both)).run()
    completion._2 foreach{_ =>
      completion._1 foreach{count =>
        println("Our flow processed " + count + " elements.")
        system.terminate()
      }
    }
  }

}

因此，我们流程的具体化价值仅在阶段完成时可用。 Flow 组件与上游源和下游接收器的具体化值无关。据我所知，只有物化器和流解释器才能获取这些物化值并适当地传播它们。

也许您的问题对无限流有一些意义，我可以想象第二个流需要在某些逻辑条件下检查嵌入在第一个流中的计数器的情况 - 但我不知道如何实现这一点。也许有人会在其他答案中解释。