使用 akka 流询问模式不起作用答案

【问题标题】：Ask pattern not working using akka streams使用 akka 流询问模式不起作用
【发布时间】：2018-11-06 04:29:08
【问题描述】：

我尝试实现一个与tutorial 略有不同的示例：

object Test extends App {
      class A extends Actor {
        override def receive: Receive = {
          case 10 => context.system.terminate()
          case x =>
            println(s"Received: $x")
            sender() ! x
        }
      }

      implicit val system = ActorSystem()
      implicit val materializer = ActorMaterializer()
      implicit val dispatcher = system.dispatcher
      implicit val askTimeout = Timeout(1, TimeUnit.SECONDS)

      val a = system.actorOf(Props[A])

      val graph = RunnableGraph.fromGraph(GraphDSL.create() {
        {
          implicit builder: GraphDSL.Builder[NotUsed] => {
            import GraphDSL.Implicits._

            val source: Source[Int, NotUsed] = Source(1 to 10)
            source ~> Flow[Int].ask(2)(a) ~> Sink.foreach[String](println)
            ClosedShape
          }
        }
      })

      val x = graph.run()
}

结果我得到了这个：

收到：1

收到：2

但我希望这样：

收到：1

1

收到：2

2

...

收到：9

9

如果我通过Source.actorRef 向演员发送消息，那么一切正常，演员会收到所有消息。似乎物化的演员在等待来自被引用演员的回应。如何解决？

【问题讨论】：

标签： scala akka akka-stream

【解决方案1】：

如果您将Sink.foreach[String](println) 替换为Sink.onComplete(println) 您将看到流程因错误而终止：

Received: 1
Received: 2
Failure(java.lang.ClassCastException: Cannot cast java.lang.Integer to scala.runtime.Nothing$)

问题在于 Source.ask 需要知道将演员的回复映射到see API docs 的类。否则它将尝试将回复强制转换为 Nothing 并在内部抛出异常。

要修复您的原始代码，请将 ask 替换为 ask[Int]：

  val graph = RunnableGraph.fromGraph(GraphDSL.create() {
    {
      implicit builder: GraphDSL.Builder[NotUsed] => {
        import GraphDSL.Implicits._

        val source: Source[Int, NotUsed] = Source(1 to 10)
        source ~> Flow[Int].ask[Int](2)(a) ~> Sink.foreach(println)
        ClosedShape
      }
    }
  })

【讨论】：