为什么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的完成消息在合并时被阻止.从文档:

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一个文件源完成后将导致合并完成.例如:

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

没有流循环

您的问题存在一个更简单的解决方案.只需使用一个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紧密结合,无论好坏.

此外,在隔离的解决方案中,循环包含在尾递归函数中,这是惯用的Scala.