Akka Stream - 並列実行可能なグラフ

Question

提案が必要です。たとえば、私は10個のグラフを作成して並列に実行するサンプルコードを作成しました。

このアプローチは正しいですか、グラフ内に複数のソースを作成して1つのグラフで並列に実行する必要がありますか？

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を、これはよさそうだ、使用して並列処理を試してみましたが、シンクはsame.Each源である以下の例では、シミュレーションであるあなたがあるように、それらを考慮すべてのそのコードは同等のものを行うには、なぜ

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() 
}