【问题标题】:Akka Streams Graph DSL notationAkka Streams 图形 DSL 表示法
【发布时间】:2016-02-29 02:42:28
【问题描述】:

我使用图形 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 => { ... 中代表什么。它只是整个形状的隐含名称吗?

【问题讨论】:

    标签: akka akka-stream


    【解决方案1】:

    GraphDSL.create 方法严重超载,可以接收输入形状数量的许多变体(包括 0)。如果您没有传入任何初始形状,则buildBlock 函数 arg 的签名(您实际定义图形构建方式的主体)如下:

    (Builder[NotUsed]) => S
    

    所以这只是一个Function1[Builder[NotUsed], S],即一个函数,它接受一个Builder[NotUsed] 的实例并返回一个Shape 实例,它是最终图。这里的NotUsedUnit 的同义词,因为您是说通过不传递任何输入份额,您不关心正在生成的输出图的具体化值。

    如果您决定传入输入形状,那么buildBlock 函数的签名会稍作更改以适应输入形状。在您的情况下,您传入 1 个输入形状,因此 buildBlock 的签名更改为:

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

    现在,这本质上是一个Function1[Builder[Mat], Function1[Graph.Shape, S]],或者一个接受Builder[Mat] 的函数(其中Mat 是输入形状的物化值类型)并返回一个接受Graph.Shape 并返回一个函数的函数S 的实例(这是一个 Shape)。

    长话短说,如果您传入形状,那么您还需要将它们声明为图形构建块函数上的绑定参数,但作为第二个输入函数(因此附加了=>)。

    【讨论】:

      猜你喜欢
      • 2017-12-23
      • 2017-09-19
      • 1970-01-01
      • 2019-04-13
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2018-02-06
      相关资源
      最近更新 更多