scala - Akka Streams 入口和导出匹配

Question

这是我能想出的使用 Partition 和 Merge 的最简单的图表，但运行时出现以下错误:要求失败:入口[]和导出[]必须对应于入口[Merge.in0，Merge.in1]和导出[Partition.out0，Partition.out1]

我知道该消息表明我的输出多于输入或未连接的流，但我似乎无法在这个简单的示例中看到不匹配的地方。

感谢任何帮助。

图表:

    def createGraph()(implicit actorSystem: ActorSystem): Graph[ClosedShape, Future[Done]] = {
      GraphDSL.create(Sink.ignore) { implicit builder: GraphDSL.Builder[Future[Done]] => s =>
          import GraphDSL.Implicits._
          val inputs: List[Int] = List(1, 2, 3, 4)
          val source: Source[Int, NotUsed] = Source(inputs)

          val messageSplit: UniformFanOutShape[Int, Int] = builder.add(Partition[Int](2, i => i%2))

          val messageMerge: UniformFanInShape[Int, Int] = builder.add(Merge[Int](2))

          val processEven: Flow[Int, Int, NotUsed] = Flow[Int].map(rc => {
            actorSystem.log.debug(s"even:  $rc")
            rc
          })

          val processOdd: Flow[Int, Int, NotUsed] = Flow[Int].map(rc => {
            actorSystem.log.debug(s"odd: $rc")
            rc
          })

          source ~> messageSplit.in
          messageSplit.out(0) -> processEven -> messageMerge.in(0)
          messageSplit.out(1) -> processOdd -> messageMerge.in(1)
          messageMerge.out ~> s
          ClosedShape
      }
    }

测试:

import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl.{Flow, GraphDSL, Merge, Partition, RunnableGraph, Sink, Source}
import akka.{Done, NotUsed}
import org.scalatest.FunSpec

import scala.concurrent.Future
class RoomITSpec extends FunSpec {

  implicit val actorSystem: ActorSystem = ActorSystem("RoomITSpec")
  implicit val actorCreator: ActorMaterializer = ActorMaterializer()
  describe("graph") {
    it("should run") {
      val graph = createGraph()
      RunnableGraph.fromGraph(graph).run
    }
  }
}

Answer 1

小句法错误。

// Notice the curly arrows
messageSplit.out(0) ~> processEven ~> messageMerge.in(0)
messageSplit.out(1) ~> processOdd ~> messageMerge.in(1)

而不是你写的:

// Straight arrows
messageSplit.out(0) -> processEven -> messageMerge.in(0)
messageSplit.out(1) -> processOdd -> messageMerge.in(1)

您最终生成(并丢弃)元组而不是添加到图中。