Akka流通过流限制并行/处理流的吞吐量

Question

我有一个用例，我想向外部系统发送一条消息，但是发送此消息的流接受并返回一种我不能在下游使用的类型。这是通过流的一个很好的用例。我正在使用实现这里。最初，我担心如果processingFlow使用mapAsyncUnordered，那么这个流就无法工作。因为处理流可能会重新排序消息，而zip可能会用不正确的对推出一个元组。例如，在下面的示例中。

  val testSource = Source(1 until 50)
  val processingFlow: Flow[Int, Int, NotUsed] = Flow[Int].mapAsyncUnordered(10)(x => Future {
    Thread.sleep(Random.nextInt(50))
    x * 10
  })
  val passThroughFlow = PassThroughFlow(processingFlow, Keep.both)

  val future = testSource.via(passThroughFlow).runWith(Sink.seq)

我希望处理流程能够根据其输入重新排序其输出，我将得到如下结果：

[(30,1), (40,2),(10,3),(10,4), ...]

在右边(传递的内容总是有序的)，但是左边，通过我的mapAsyncUnordered，可能会加入一个不正确的元素，从而形成一个糟糕的元组。

实际上，我得到的是：

[(10,1), (20,2),(30,3),(40,4), ...]

每次都是这样。经过进一步的研究，我注意到代码运行缓慢，实际上它根本没有并行运行，尽管我的map异步无序。我尝试在前后引入一个缓冲区，以及一个异步边界，但它似乎总是按顺序运行。这解释了为什么它总是有序的，但我希望我的处理流程具有更高的吞吐量。

我做了以下的工作：

object PassThroughFlow {

  def keepRight[A, A1](processingFlow: Flow[A, A1, NotUsed]): Flow[A, A, NotUsed] =
    keepBoth[A, A1](processingFlow).map(_._2)

  def keepBoth[A, A1](processingFlow: Flow[A, A1, NotUsed]): Flow[A, (A1, A), NotUsed] =
    Flow.fromGraph(GraphDSL.create() { implicit builder => {
      import GraphDSL.Implicits._

      val broadcast = builder.add(Broadcast[A](2))
      val zip = builder.add(ZipWith[A1, A, (A1, A)]((left, right) => (left, right)))

      broadcast.out(0) ~> processingFlow ~> zip.in0
      broadcast.out(1) ~> zip.in1

      FlowShape(broadcast.in, zip.out)
    }
    })
}

object ParallelPassThroughFlow {


  def keepRight[A, A1](parallelism: Int, processingFlow: Flow[A, A1, NotUsed]): Flow[A, A, NotUsed] =
    keepBoth(parallelism, processingFlow).map(_._2)

  def keepBoth[A, A1](parallelism: Int, processingFlow: Flow[A, A1, NotUsed]): Flow[A, (A1, A), NotUsed] = {
    Flow.fromGraph(GraphDSL.create() { implicit builder =>
      import GraphDSL.Implicits._

      val fanOut = builder.add(Balance[A](outputPorts = parallelism))
      val merger = builder.add(Merge[(A1, A)](inputPorts = parallelism, eagerComplete = false))

      Range(0, parallelism).foreach { n =>
        val passThrough = PassThroughFlow.keepBoth(processingFlow)
        fanOut.out(n) ~> passThrough ~> merger.in(n)
      }

      FlowShape(fanOut.in, merger.out)
    })
  }

}

两个问题：

1. 在原始实施中，为什么通过流中的zip限制映射异步无序的并行性？
2. 我的作品是声音的，还是可以改进的？我基本上把我的输入输入到多个通过流的堆栈中，并将它们合并起来。它似乎有我想要的属性(并行但保持顺序，即使处理流程重新排序)，但是有些东西感觉不对

Answer 1

您所看到的行为是broadcast和zip如何工作的结果：broadcast在其所有输出信号需求时向下游发出信号；zip在信令需求之前等待其所有输入(并向下游发射)。

broadcast.out(0) ~> processingFlow ~> zip.in0
broadcast.out(1) ~> zip.in1

考虑第一个元素(1)通过上面的图的移动。1广播给processingFlow和zip。zip立即接收到它的一个输入(1)，并等待它的另一个输入(10)，这将需要稍长的时间才能到达。只有当zip同时获得1和10时，它才会从上游提取更多的元素，从而触发第二个元素(2)在流中的移动。诸若此类。

至于你的ParallelPassThroughFlow，我不知道为什么你“感觉不对劲”。