最近在项目中需要实现图的一些操作，因此，初步考虑使用Akka Stream的Graph实现。从而学习了下：
一、介绍
我们知道在Akka Stream中有三种简单的线性数据流操作：Source/Flow/Sink。但是当我们需要使用一些复杂的操作，例如扇入和扇出时，可能就需要使用图相关的流操作了。因此，我们可以这样认为，Akka Stream的Graph是一种运算方案，他可能是简单的线性数据流，也可以由基础的流图组合而成的复杂的数据流程。因为Graph只是对数据流运算的简单描述，所以它是可以重复利用的。
二、依赖
要使用Akka Stream的Graph，我们需要添加下面的依赖：
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-stream_2.12</artifactId>
<version>2.5.18</version>
</dependency>三、构建Graph
Graph是由简单的Flow组成的，这些Flow用作图形中的线性连接以及用作Flow的扇入和扇出点的连接点。Akka Stream目前提供了下面这些连接点：
1、扇出：
（1）Broadcast[T]：（1输入，N输出）给定输入元件发射到每个输出
（2）Balance[T]：（1输入，N输出）给定输入元件发射到其输出端口之一
（3）UnzipWith[In,A,B,...]：（1个输入，N个输出）采用1个输入的函数，给定每个输入的值发出N个输出元素（其中N <= 20）
（4）UnZip[A,B]：（1个输入，2个输出）将元组流(A,B)拆分为两个流，一个是类型A，另一个是类型B
2、扇入：
（1）Merge[In]：（N个输入，1个输出）从输入中随机选取将它们逐个推入其输出
（2）MergePreferred[In]：Merge但是如果元素在最受欢迎的端口上可用，它会从中选择，否则从中随机从其他端口上选
（3）MergePrioritized[In]：Merge但是如果元素在所有输入端口上都可用，它会根据它们的优先级随机选择它们
（4）MergeLatest[In]：（N个输入，1个输出）发出List[In]，当第i个输入流发出元素时，发出的列表中的第i个元素被更新
（5）ZipWith[A,B,...,Out]：（N个输入，1个输出），其取N个输入的函数，给出每个输入的值，发出1个输出元素
（6）Zip[A,B]：（2个输入，1个输出）是一个ZipWith专用于压缩和解的输入流A和B成元组流(A,B)
（7）Concat[A]：（2个输入，1个输出）连接两个流（首先消耗一个，然后消耗第二个）
四、例子
现在假设我们需要实现如下图所示的一个Graph
我们可以用akka-stream提供的GraphDSL来构建Graph。GraphDSL继承了GraphApply的create方法，GraphDSL.create(...)就是构建Graph的方法，因此，我们可以使用如下代码创建上图所示的Graph：
val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
import GraphDSL.Implicits._
val in = Source(1 to 10)
val out = Sink.ignore
val bcast = builder.add(Broadcast[Int](2))
val merge = builder.add(Merge[Int](2))
val f1, f2, f3, f4 = Flow[Int].map(_ + 10)
in ~> f1 ~> bcast ~> f2 ~> merge ~> f3 ~> out
bcast ~> f4 ~> merge
ClosedShape
})注意：在这个里面我们需要引入import GraphDSL.Implicits._。是为了将~>（读作边缘，通过或者到），以及他的相反操作<~引入到代码的范围内。
原文地址：https://www.cnblogs.com/junjiang3/p/10016003.html