Source.Single() 过早终止具有 2 个源的流答案

【问题标题】：Source.Single() terminates stream with 2 sources prematurelySource.Single() 过早终止具有 2 个源的流
【发布时间】：2019-11-22 01:34:57
【问题描述】：

我已经定义了一个简单的图表，它将恒定流（通过Source.Single(5000) 定义）与非恒定源（例如Source(1 to 100)）相结合。我的图表通过ZipLatestWith 运算符将这两个数字相加。目的是使流的输出为5001, 5002, 5003, 5004, ..., 5100。实际上，程序的输出只是5000，然后流可能会终止，因为Single 源已完成。

如何获得预期的结果，其中恒定值5000 的源与非常量源的每个值相结合？请注意，由于概念上的原因（与此特定示例无关），常量源仍然是流的源，这一点很重要。完整的代码示例如下，简单的将数字5000打印到控制台。

Main.scala：

import akka.NotUsed
import akka.actor.typed.scaladsl.Behaviors
import akka.actor.typed.{ActorSystem, Behavior, Terminated}
import akka.stream.scaladsl.{Flow, GraphDSL, Sink, Source, ZipLatestWith}
import akka.stream.{FlowShape, Materializer}

object Main {
  def apply(): Behavior[NotUsed] = {
    Behaviors.setup { context =>

      val constantStream: Source[Int, NotUsed] = Source.single(5000)
      val graph: Flow[Int, Int, NotUsed] = Flow.fromGraph(GraphDSL.create(){
        implicit builder =>
          import GraphDSL.Implicits._

          val zipper = builder.add(ZipLatestWith[Int, Int, Int]((a: Int, b: Int) => a * b))
          constantStream ~> zipper.in1

          FlowShape(zipper.in0, zipper.out)
      })

      Source(1 to 100)
        .via(graph)
        .to(Sink.foreach(println)).run()(Materializer.apply(context))

      Behaviors.receiveSignal {
        case (_, Terminated(_)) =>
          Behaviors.stopped
      }
    }
  }

  def main(args: Array[String]): Unit = {
    ActorSystem(Main(), "test")
  }
}

build.sbt:

scalaVersion := "2.12.6"
libraryDependencies += "com.typesafe.akka" %% "akka-actor-typed" % "2.6.0"
libraryDependencies += "com.typesafe.akka" %% "akka-stream-typed" % "2.6.0"
libraryDependencies += "ch.qos.logback" % "logback-classic" % "1.2.3"

【问题讨论】：

标签： scala akka akka-stream

【解决方案1】：

只需改用Source.repeat(5000)。

【讨论】：

【解决方案2】：

除了@cbley 所说的，如果可能的话，你应该尽量避免GraphDSL。最好使用现有的组合器，因为它不易出错且更易于理解。

上图可以用

表示

Source(1 to 100).zip(Source.repeat(5000))
  .map { case (x, y) => x + y }
  .to(Sink.foreach(println))

【讨论】：