Akka 流到演员的问题答案

【问题标题】：Akka stream to actor issueAkka 流到演员的问题
【发布时间】：2018-01-25 16:27:21
【问题描述】：

我正在尝试使用 Sink.actorRefWithAck 将 http 响应流式传输到演员，但不知何故，除了 init 和 complete msgs 之外，没有任何东西发送到这个 sink actor。不过，我可能遗漏了一些非常明显的东西。接收器参与者正在启动。以下是流演员的代码：

override def receive: Receive = {
    case GetTestData(p, id) =>
      // Get the data and pipes it to itself through a message as recommended
  
      http.singleRequest(HttpRequest(uri = uri.format(p, id)))
        .pipeTo(self)

    case HttpResponse(StatusCodes.OK, _, entity, _) =>
      // Sink as actorRef
      val sink = Sink.actorRefWithAck(context.actorOf(TestJob2.props(), "testJob2Actor"), Init, Ack, Complete)
      // Pipe the response body to the actor sink
      entity.dataBytes.runWith(sink)

    case resp @ HttpResponse(code, _, _, _) =>
      log.error("Request to test job failed, response code: " + code)
      // Discard the flow to avoid backpressure
      resp.discardEntityBytes()

    case _ => log.warning("Unexpected message in TestJobActor")
}

这里是 sink actor 的代码：

override def receive: Receive = {
  case Init =>
    log.info("TestJob2Actor got init sink message")
    sender ! Ack

  case Complete => log.info("TestJob2Actor got complete sink message")

  case b: ByteString =>
    log.debug(b.utf8String)
    sender ! Ack

  case _ => log.warning("Unexpected message in TestJob2Actor")
}

输出：

2018-01-25 17:26:58,530 INFO com.mcma.actors.Supervisor akka.tcp://alor-system@10.33.135.82:8000/user/supervisorActor - SupervisorActor 将 GetTestData 消息转发给 TestJobActor

2018-01-25 17:26:59,243 INFO com.mcma.actors.jobs.TestJob akka.tcp://alor-system@10.33.135.82:8000/user/supervisorActor/testJobActor - TestJob 演员开始

2018-01-25 17:27:00,052 INFO com.mcma.actors.jobs.TestJob2 akka.tcp://alor-system@10.33.135.82:8000/user/supervisorActor/testJobActor/testJob2Actor - TestJob2 演员开始

2018-01-25 17:27:00,067 INFO com.mcma.actors.jobs.TestJob2 akka.tcp://alor-system@10.33.135.82:8000/user/supervisorActor/testJobActor/testJob2Actor - TestJob2Actor 得到了初始化接收消息

2018-01-25 17:27:00,083 INFO com.mcma.actors.jobs.TestJob2 akka.tcp://alor-system@10.33.135.82:8000/user/supervisorActor/testJobActor/testJob2Actor - TestJob2Actor 完成接收消息

【问题讨论】：

标签： scala akka akka-stream

【解决方案1】：

想法：（1）来自entity.dataBytes 的Source 可能是空的，或者（2）接收器参与者实际上正在处理ByteString 块，但是日志记录级别使得调试消息不可见。尝试将日志记录级别设置为 DEBUG，或将 log.debug(b.utf8String) 更改为 log.info(b.utf8String)。

【讨论】：

问题，但我只有一个日志行，一次包含整个 json 消息，这非常小，但我不明白为什么如果我有字节流，它不按字节记录？
要将Source 分块为有效的JSON ByteString 段，您可以使用JsonFraming: entity.dataBytes.via(JsonFraming.objectScanner(Int.MaxValue)).runWith(sink)
谢谢，它正在正确地分块 json 对象并将消息一一发送给参与者。也就是说，系统的目标只是充当管道：接收大的 http 响应并通过接收器 actor 中的 http 流回数据流，最好的方法性能是什么？因为我认为如果我对收到的每条消息/json obj 发出 http 请求，atm 流的概念就会被打破，对吗？是否有可能使用标头“Transfer-Encoding：chunked”标头来维护http连接？