Akka Streams Graph DSL 表示法

Question

我使用 graph dsl 根据我看到的一些示例代码创建了一些流处理作业。一切运行良好，我只是无法理解符号：（针对 2.4 更新）

def elements: Source[Foos] = ...
def logEveryNSink = // a sink that logs
def cleaner: Flow[Foos, Bars, Unit] = ...

def boolChecker(bar: Bar)(implicit ex: ExecutionContext): Future[Boolean] = ...

val mySink = Sink.foreach[Boolean](println(_))

val lastly = Flow[Bars].mapAsync(2)(x => boolChecker(x).toMat(mySink)(Keep.right)

val materialized = RunnableGraph.fromGraph(
 GraphDSL.create(lastly) { implicit builder =>
  baz => {
   import GraphDSL.Implicits._
   val broadcast1 = builder.add(Broadcast[Foos](2))
   val broadcast2 = builder.add(Broadcast[Bars](2))
   elements ~> broadcast1 ~> logEveryNSink(1)
               broadcast1 ~> cleaner ~> broadcast2 ~> baz
                                     ~> broadcast2 ~> logEveryNSink(1)
   ClosedShape
 }
}
).run()

我理解包含的隐式构建器，但我不确定 baz 在 { implicit builder => baz => { ... 中代表什么。它只是整个形状的隐式名称吗？

Answer 1

GraphDSL.create 方法重载以接受输入形状数量的许多变体（包括 0）。如果您没有传入任何初始形状，那么 buildBlock 函数 arg 的签名（您实际定义图形构建方式的主体）如下所示：

(Builder[NotUsed]) => S

所以这只是一个 Function1[Builder[NotUsed], S]，也就是说，一个函数接受一个 Builder[NotUsed] 的实例和 returns 一个 Shape 实例，这是最终的图形.这里的 NotUsed 是 Unit 的同义词，因为你是说不传递任何输入共享，你不关心正在生成的输出图的物化值。

如果您决定传入输入形状，那么 buildBlock 函数的签名会稍作更改以适应输入形状。在您的例子中，您传入了 1 个输入形状，因此 buildBlock 的签名更改为：

(Builder[Mat]) => Graph.Shape => S

现在，这本质上是一个 Function1[Builder[Mat], Function1[Graph.Shape, S]]，或者一个接受 Builder[Mat] 的函数（其中 Mat 是输入形状的物化值类型）和 returns 一个函数，它接受 Graph.Shape 和 returns 一个 S 的实例（这是一个 Shape）。

长话短说，如果你传入形状，那么你还需要将它们声明为图形构建块函数的绑定参数，但作为第二个输入函数（因此需要额外的 =>）。

Akka Streams Graph DSL 表示法

Akka Streams Graph DSL notation

akka

akka-stream