scala - 如何有条件地将消息传递到特定的下一个阶段(以及其他阶段)？

Question

考虑一个简单的场景，根据传递的消息的某些属性，我希望它由特定的下一阶段处理并继续。

[Source[ActionMessage]] ~> [Flow[ActionMessage, EnrichedActionMessage]] 
~> (eAM: EnrichedActionMessage => eAM.actionType match {
      case ActionA => eAM ~> Flow[EnrichedActionMessage, ReactionA] ~> Sink[ReactionA]
      case ActionB => eAM ~> Flow[EnrichedActionMessage, ReactionB] ~> Sink[ReactionB]
      case ActionC => eAM ~> Flow[EnrichedActionMessage, ReactionC] ~> Sink[ReactionC]
    })

如何实现到阶段图阶段的条件路由？

Answer 1

此答案基于akka-stream版本2.4.2-RC1。其他版本中的 API 可能略有不同。该依赖项可以由 sbt 消耗:

libraryDependencies += "com.typesafe.akka" %% "akka-stream" % "2.4.2-RC1"

---

使用分区组件:

val shape = GraphDSL.create() { implicit b ⇒
  import GraphDSL.Implicits._

  val first = b.add(Sink.foreach[Int](elem ⇒ println("first:\t" + elem)))
  val second = b.add(Sink.foreach[Int](elem ⇒ println("second:\t" + elem)))
  val third = b.add(Sink.foreach[Int](elem ⇒ println("third:\t" + elem)))
  val p = b.add(Partition[Int](3, elem ⇒ elem match {
    case 0                ⇒ 0
    case elem if elem < 0 ⇒ 1
    case elem if elem > 0 ⇒ 2
  }))

  p ~> first
  p ~> second
  p ~> third

  SinkShape(p.in)
}
Source(List(0, 1, 2, -1, 1, -5, 0)).to(shape).run()

/*
Output:
first: 0
third: 1
third: 2
second: -1
third: 1
second: -5
first: 0
*/

除了SinkShape，您还可以返回一个new FanOutShape3(p.in, p.out(0), p.out(1), p.out(2)) 如果您希望稍后对元素进行任何处理。