Akka Stream：在图形构建器阶段动态增长出口答案

【问题标题】：Akka Stream : dynamically grow outlets during graph builder phaseAkka Stream：在图形构建器阶段动态增长出口
【发布时间】：2016-09-10 02:34:37
【问题描述】：

我了解在 Akka Stream 中，一个出口必须连接到一个入口，并且没有自动支持将多个接收器连接到同一个源。所以必须插入Broadcast等中间对象。

我正在将信号处理 DAG 转换为 Akka 流图，如果我可以在遍历中发现源时动态地将接收器添加到源中，这将对我有很大帮助。如果我有一个自定义的GraphStage，我可以拥有我自己的Shape，其outlets 集合在Graph.create 阶段动态增长吗？正常的 DSL 操作 ~> 由此调用支持：

b.addEdge(importAndGetPort(b), to)

构建者如何在此处“获取”Outlet，我能否按需扩展我的形状？

如果这不起作用，是否可以“弹出”先前的广播，断开其边缘并在图形构建期间将它们与新的更大的广播连接起来？

【问题讨论】：

我不知道潜在问题的答案，但这种设计有一个很大的缺点：源只会与最慢的目的地一样快地广播值。只有当所有接收器都处理完一条消息时，广播才会发出更多的需求信号。您的问题似乎更适合直接使用 Actors 。然后，您可以使用路由将目的地动态添加到 Actor。 doc.akka.io/docs/akka/2.4.2/scala/routing.html
@RamonJRomeroyVigil 是的，只有在所有接收器都准备好后，源才能处理。这是有意的，因为源节点可能会执行昂贵且不可重复的计算。如果需要，我可以在源和接收器之间插入缓冲区。

标签： scala akka-stream

【解决方案1】：

GraphDSL 不允许动态改变你的形状。

但是，从 Akka 2.4.10 开始，您可以使用 BroadcastHub（和 MergeHub）。

BroadcastHub 可以为您提供一个具体化为 Source 的 Sink。反过来，这个 Source 可以根据需要多次具体化，以动态附加多个订阅者。

所以对于你的 DAG 的一个节点（例如 indegree=1 和 outdegree=3），你可以有类似的东西

val hubSource = inEdgeSource.toMat(BroadcastHub.sink(bufferSize = ...))(Keep.right).run()

val nodeSink1 = hubSource.to(outEdgeSink1).run()
val nodeSink2 = hubSource.to(outEdgeSink2).run()
val nodeSink3 = hubSource.to(outEdgeSink3).run()

Akka 文档：

http://doc.akka.io/docs/akka/2.4/scala/stream/stream-dynamic.html#Dynamic_fan-in_and_fan-out_with_MergeHub_and_BroadcastHub

【讨论】：