为什么 Akka 流循环没有在此图中结束？

Question

我想创建一个在下沉之前循环 n 次的图表。我刚刚创建了这个满足我要求的示例，但在下沉后并没有结束，我真的不明白为什么。有人可以启发我吗？

谢谢。

    import akka.actor.ActorSystem
    import akka.stream.scaladsl._
    import akka.stream.{ActorMaterializer, UniformFanOutShape}

    import scala.concurrent.Future

    object test {
      def main(args: Array[String]) {
        val ignore: Sink[Any, Future[Unit]] = Sink.ignore
        val closed: RunnableGraph[Future[Unit]] = FlowGraph.closed(ignore) { implicit b =>
          sink => {
            import FlowGraph.Implicits._

            val fileSource = Source.single((0, Array[String]()))
            val merge = b.add(MergePreferred[(Int, Array[String])](1).named("merge"))
            val afterMerge = Flow[(Int, Array[String])].map {
              e =>
                println("after merge")
                e
            }
            val broadcastArray: UniformFanOutShape[(Int, Array[String]), (Int, Array[String])] = b.add(Broadcast[(Int, Array[String])](2).named("broadcastArray"))
            val toRetry = Flow[(Int, Array[String])].filter {
              case (r, s) => {
                println("retry " + (r < 3) + " " + r)
                r < 3
              }
            }.map {
              case (r, s) => (r + 1, s)
            }
            val toSink = Flow[(Int, Array[String])].filter {
              case (r, s) => {
                println("sink " + (r >= 3) + " " + r)
                r >= 3
              }
            }
            merge.preferred <~ toRetry <~ broadcastArray
            fileSource ~> merge ~> afterMerge ~> broadcastArray ~> toSink ~> sink
          }
        }
        implicit val system = ActorSystem()
        implicit val _ = ActorMaterializer()
        val run: Future[Unit] = closed.run()
        import system.dispatcher
        run.onComplete {
          case _ => {
            println("finished")
            system.shutdown()
          }
        }
      }
    }`

Answer 1

Stream 永远不会完成，因为合并永远不会发出完成信号。

格式化你的图形结构后，它基本上看起来像：

//ignoring the preferred which is inconsequential

fileSource ~> merge ~> afterMerge ~> broadcastArray ~> toSink ~> sink
              merge <~ toRetry    <~ broadcastArray

未完成的问题根源于您的合并步骤：

// 2 inputs into merge

fileSource ~> merge 
              merge <~ toRetry

一旦 fileSource 发出了它的单个元素（即(0, Array.empty[String])），它就会发送一个complete 消息进行合并。

但是，fileSource 的完成消息在合并时被阻止。来自documentation：

akka.stream.scaladsl.MergePreferred

当所有上游完成 (eagerClose=false) 或一个时完成 上游完成（eagerClose=true）

在 所有 输入流完成之前，合并不会发送 complete。

// fileSource is complete ~> merge 
//                           merge <~ toRetry is still running

// complete fileSource + still running toRetry = still running merge

因此，合并将等到toRetry 也完成。但是toRetry 永远不会完成，因为它正在等待merge 完成。

如果您希望在 fileSource 完成后完成特定图表，则只需设置 eagerClose=True，这将导致在 fileSource 完成后完成合并。例如：

//Add this true                                             |
//                                                          V
val merge = b.add(MergePreferred[(Int, Array[String])](1, true).named("merge")

没有流循环

针对您的问题有一个更简单的解决方案。只需使用一个使用tail recursive 函数的Flow.map 阶段：

//Note: there is no use of akka in this implementation

type FileInputType = (Int, Array[String])

@scala.annotation.tailrec
def recursiveRetry(fileInput : FileInputType) : FileInputType = 
  fileInput match { 
    case (r,_) if r >= 3  => fileInput
    case (r,a)            => recursiveRetry((r+1, a))
  }    

然后您的流将减少到

//ring-fenced akka code

val recursiveRetryFlow = Flow[FileInputType] map recursiveRetry

fileSource ~> recursiveRetryFlow ~> toSink ~> sink

结果是一个更干净的流，它避免了将“业务逻辑”与 akka 代码混合。这允许完全独立于任何第三方库的重试功能的单元测试。您在流中嵌入的重试循环是“业务逻辑”。因此，无论好坏，混合实现都与 akka 紧密耦合。

此外，在隔离解决方案中，循环包含在尾递归函数中，即idiomatic Scala。