Akka Stream - 并行可运行图

Question

需要建议，我需要并行运行多个源图，例如，我创建了这个示例代码，我正在创建 10 个图并并行运行它们。

这是正确的方法还是我应该在一个图中创建多个源并在一个图中并行运行它们？

def createGraph(start: Int, end: Int, name: String) = {
  RunnableGraph.fromGraph(GraphDSL.create() {
    implicit builder =>
      import GraphDSL.Implicits._
      val s = Source(start to end)
      val f = Flow[Int].map[String](x => x.toString)
      val sink = Sink.foreach[String](x => println(name + ":" + x))

      val t = builder.add(s)

      val flow1 = builder.add(f)

      t ~> flow1 ~> sink

      ClosedShape
  })
}


(1 to 10).map(x => createGraph(x, x + 10, "g" + x)).map(_.run())

谢谢 阿伦

Answer 1

我尝试使用 http://doc.akka.io/docs/akka/2.4.4/scala/stream/stream-parallelism.html 进行并行处理，这看起来不错，其中我的源不同但流和接收器相同。在下面的示例中，每个源都是模拟，将它们视为您正在从某个外部源作为流读取：

object TestParallelGraph extends App {

  implicit val system = ActorSystem("test")
  implicit val dispacher = system.dispatcher
  implicit val materializer = ActorMaterializer()

  val listOfDifferentSource=List(1,2,3) //consider we have to read data from various sources


 def createGraph() = {
    RunnableGraph.fromGraph(GraphDSL.create() {
      implicit builder =>
        import GraphDSL.Implicits._

        val merge=builder.add(Merge[Int](listOfDifferentSource.length))

        val flow=builder.add(Flow[Int].map(_ + 10)) //just random flow to add 10

        //again as mentioned above creating source with different information to simulate
        Source(listOfDifferentSource.head*100 to 100* listOfDifferentSource.head+10) ~> merge ~> flow ~> Sink.foreach(println)

        for{
          i <- listOfDifferentSource.tail //for each other source
        }yield (Source(100*i to 100*i+10) ~> merge) 

        ClosedShape
    })
  }

  createGraph().run()
}