Akka Stream之Graph

Posted 2021-01-22 junjiang3

最近在项目中需要实现图的一些操作，因此，初步考虑使用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._。是为了将~>（读作边缘，通过或者到），以及他的相反操作<~引入到代码的范围内。