我試圖將 AkkaSource分成兩個獨立的。
val requestFlow = Flow[BodyPartEntity].to(Sink.seq) // convert to Seq[BodyPartEntity]
val dataFlow = Flow[BodyPartEntity].to(Sink.asPublisher(fanout = false)) // convert to Publisher[BodyPartEntity]
implicit class EitherSourceExtension[L, R, Mat](source: Source[FormData.BodyPart, Mat]) {
def partition(left: Sink[BodyPartEntity, NotUsed], right: Sink[BodyPartEntity, NotUsed]): Graph[ClosedShape, NotUsed] = {
GraphDSL.create() { implicit builder =>
import akka.stream.scaladsl.GraphDSL.Implicits._
val partition = builder.add(Partition[FormData.BodyPart](2, element => if (element.getName == "request") 0 else 1))
source ~> partition.in
partition.out(0).map(_.getEntity) ~> left
partition.out(1).map(_.getEntity) ~> right
ClosedShape
}
}
}
如何轉換requestFlow成Seq[BodyPartEntity]和dataFlow成Publisher[BodyPartEntity]
uj5u.com熱心網友回復：
您可以為此使用BroadcastHub。來自檔案：
BroadcastHub 可用于由一組動態消費者使用來自公共生產者的元素。
簡化代碼：
val runnableGraph: RunnableGraph[Source[Int, NotUsed]] =
Source(1 to 5).toMat(
BroadcastHub.sink(bufferSize = 4))(Keep.right)
val fromProducer: Source[Int, NotUsed] = runnableGraph.run()
// Process the messages from the producer in two independent consumers
fromProducer.runForeach(msg => println("consumer1: " msg))
fromProducer.runForeach(msg => println("consumer2: " msg))
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/372711.html
上一篇：呼叫rpc方法之間的功能