如何从官方文档中解释这个 Akka Streams 图？

Question

我对@9​​87654321@ 官方托管的这个示例代码有几个问题：

val topHeadSink = Sink.head[Int]
val bottomHeadSink = Sink.head[Int]
val sharedDoubler = Flow[Int].map(_ * 2)

RunnableGraph.fromGraph(GraphDSL.create(topHeadSink, bottomHeadSink)((_, _)) { implicit builder =>
  (topHS, bottomHS) =>
  import GraphDSL.Implicits._
  val broadcast = builder.add(Broadcast[Int](2))
  Source.single(1) ~> broadcast.in

  broadcast.out(0) ~> sharedDoubler ~> topHS.in
  broadcast.out(1) ~> sharedDoubler ~> bottomHS.in
  ClosedShape
})

1. 您何时通过create 传递图表？

为什么topHeadSink, bottomHeadSink是通过create传入的，而sharedDoubler不是？它们有什么区别？

1. 什么时候需要builder.add？

我可以在没有builder.add 的情况下在图表外创建广播吗？如果我在图表中添加几个流，我是否也应该通过builder.add 添加流？有时我们需要 builder.add 而有时不需要，这非常令人困惑。

更新

我觉得这仍然令人困惑：

这些方法之间的区别在于，使用builder.add(...) 导入会忽略导入图的物化值，而通过工厂方法导入则允许包含它。

topHS, bottomHS 是从create 导入的，因此它们将保留其物化值。如果我做builder.add(topHS)怎么办？

你如何解释sharedDoubler：它是否具有物化价值？如果我使用builder.add 怎么办？

1. ((_,_)) 的 GraphDSL.create(topHeadSink, bottomHeadSink)((_, _)) 是什么意思？

它看起来像我们只需要的样板，但我不确定它是什么。

Answer 1

1. 什么时候通过 create 传入图表？

当您想要获取传递给create 工厂方法的图形的具体化值时。你问题中RunnableGraph的类型是RunnableGraph[(Future[Int], Future[Int])]，意思是图的物化值是(Future[Int], Future[Int])：

val g = RunnableGraph.fromGraph(...).run() // (Future[Int], Future[Int])
val topHeadSinkResult    = g._1 // Future[Int]
val bottomHeadSinkResult = g._2 // Future[Int]

现在考虑以下变体，它在图形“内部”定义汇并丢弃物化值：

val g2 = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder =>
  import GraphDSL.Implicits._

  val topHeadSink = Sink.head[Int]
  val bottomHeadSink = Sink.head[Int]
  val broadcast = builder.add(Broadcast[Int](2))

  Source.single(1) ~> broadcast.in
  broadcast.out(0) ~> sharedDoubler ~> topHeadSink
  broadcast.out(1) ~> sharedDoubler ~> bottomHeadSink
  ClosedShape
}).run() // NotUsed

g2 的值为NotUsed。

1. 什么时候需要builder.add？

图的所有组件都必须添加到构建器中，但是~> 运算符的变体可以将最常用的组件（例如Source 和Flow）添加到构建器中在被子下。但是，如果您使用图 DSL。

请注意，对于简单的图表，您可以使用联结而不必使用 Graph DSL。这是来自documentation 的示例：

val sendRmotely = Sink.actorRef(actorRef, "Done")
val localProcessing = Sink.foreach[Int](_ => /* do something usefull */ ())

val sink = Sink.combine(sendRmotely, localProcessing)(Broadcast[Int](_))

Source(List(0, 1, 2)).runWith(sink)

1. 这是什么意思？ ((_,_)) 的 GraphDSL.create(topHeadSink, bottomHeadSink)((_, _))?

这是一个柯里化参数，用于指定要保留的具体化值。此处使用((_, _)) 等同于：

val g = RunnableGraph.fromGraph(GraphDSL.create(topHeadSink, bottomHeadSink)((t, b) => (t, b)) {
  implicit builder => (topHS, bottomHS) =>
  ...
}).run() // (Future[Int], Future[Int])

换句话说，((_, _)) 在此上下文中是 ((t, b) => (t, b)) 的简写，它保留了传入的两个接收器的相应物化值。例如，如果您只想保留 @ 的物化值987654344@，您可以将呼叫更改为：

val g = RunnableGraph.fromGraph(GraphDSL.create(topHeadSink, bottomHeadSink)((t, _) => t) {
  implicit builder => (topHS, bottomHS) =>
  ...
}).run() // Future[Int]