是否可以使用 Akka-stream 创建一个可以在 2 个不同内部形状之间切换的流？答案

【问题标题】：Is it possible to create a Flow with Akka-stream that can switch between 2 different inner Shapes?是否可以使用 Akka-stream 创建一个可以在 2 个不同内部形状之间切换的流？
【发布时间】：2020-12-03 13:47:49
【问题描述】：

我想要一个复杂的Flow，我可以根据流入图表的数据在两个不同的Shapes 之间切换。当我们返回 ClosedShape 时，图形是静态的，但是当我们返回 FlowShape 时，我想知道是否可以在其中创建某种动态流。我正在查看this question，似乎他们使用了Partition，我不知道如何应用或者它是否真的解决了我的问题。

我从这个例子开始，我被代码中的注释卡住了。

import akka.actor.ActorSystem
import akka.stream.FlowShape
import akka.stream.scaladsl.{Flow, GraphDSL, Sink, Source}

import scala.concurrent.duration._

object StreamOpenGraphsWithMultipleFlows extends App {

  run()

  def run() = {
    implicit val system = ActorSystem("StreamOpenGraphsWithMultipleFlows")

    val fastSource = Source(1 to 1000).throttle(50, 1 second)
    val slowSource = Source(1 to 1000).throttle(5, 1 second)
    val INC = 5
    val MULTI = 10
    val DIVIDE = 2

    val incrementer = Flow[Int].map { x =>
      val result = x + INC
      print(s" | incrementing $x + $INC -> $result")
      result
    }
    val multiplier = Flow[Int].map { x =>
      val result = x * MULTI
      print(s" | multiplying $x * $MULTI -> $result")
      result
    }
    val divider = Flow[Int].map { x =>
      val result = x / DIVIDE
      print(s" | dividing $x / $DIVIDE -> $result")
      result
    }

    def isMultipleOf(value: Int, multiple: Int): Boolean = (value % multiple) == 0

    // Step 1 - setting up the fundamental for a stream graph
    val complexFlowIncrementer = Flow.fromGraph(
      GraphDSL.create() { implicit builder =>
        import GraphDSL.Implicits._
        // Step 2 - add necessary components of this graph
        val incrementerShape = builder.add(incrementer)
        val multiplierShape = builder.add(multiplier)

        // println(s"builder.materializedValue: ${builder.materializedValue}")

        // Step 3 - tying up the components
        incrementerShape ~> multiplierShape
        // BUT I WOULD LIKE TO DO SOMETHING AS BELOW
        // if (isMultipleOf(value???, 10)) incrementerShape ~> divider
        // else incrementerShape ~> multiplierShape

        // Step 4 - return the shape
        FlowShape(incrementerShape.in, multiplierShape.out)
      }
    )
    // run the graph and materialize it
    val graph = slowSource
      .via(complexFlowIncrementer)
      .to(Sink.foreach(x => println(s" | result: $x")))
    graph.run()
  }
}

【问题讨论】：

是的。您可以编写一个分区器，该分区器对分区进行处理，然后连接到内部形状......然后内部形状连接到一个合并，该合并用作您的流程的输出。

标签： scala akka akka-stream

【解决方案1】：

这个blog post 展示了如何实现这一目标的代码示例，所以在你的情况下，你需要沿着这些思路：

val complexFlowIncrementer = Flow.fromGraph(
      GraphDSL.create() { implicit builder =>
        import GraphDSL.Implicits._
        // Step 2 - add necessary components of this graph
        val incrementerShape = builder.add(incrementer)
        val multiplierShape = builder.add(multiplier)
        val dividerShape = builder.add(divider)
        
        //add partition and merge
        val partition = builder.add(Partition[Int](2, if(isMultipleOf(_, 10)) 0 else 1)
        val merge = builder.add(Merge[Int](2))

        // println(s"builder.materializedValue: ${builder.materializedValue}")

        // Step 3 - tying up the components
        incrementerShape ~> partition
        partition.out(0) ~> dividerShape ~> merge.in(0)
        partition.out(1) ~> multiplierShape ~> merge.in(1)
       
        // Step 4 - return the shape
        FlowShape(incrementerShape.in, merge.out)
      }
    )

【讨论】：