如何在 akka actor 中使用 akka-http cachedHostConnectionPool？答案

【问题标题】：How to use the akka-http cachedHostConnectionPool inside an akka actor?如何在 akka actor 中使用 akka-http cachedHostConnectionPool？
【发布时间】：2016-02-04 20:56:45
【问题描述】：

akka http 文档在 high level client request API documentation 中提到，我们不应该在 Actor 的 Future 中使用访问 Actor 状态表单。

相反，这应该是使用的模式：

import akka.actor.Actor
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.stream.scaladsl.ImplicitMaterializer

class Myself extends Actor
  with ImplicitMaterializer
  with ActorLogging {

  import akka.pattern.pipe
  import context.dispatcher

  val http = Http(context.system)

  override def preStart() = {
    http.singleRequest(HttpRequest(uri = "http://akka.io"))
      .pipeTo(self)
  }

  def receive = {
    case HttpResponse(StatusCodes.OK, headers, entity, _) =>
      log.info("Got response, body: " + entity.dataBytes.runFold(ByteString(""))(_ ++ _))
    case HttpResponse(code, _, _, _) =>
      log.info("Request failed, response code: " + code)
  }

}

在使用cachedHostConnectionPool时我们应该做类似的事情吗？

例如：

import akka.actor.Actor
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.stream.scaladsl.ImplicitMaterializer

class Myself extends Actor
  with ImplicitMaterializer
  with ActorLogging {

  import akka.pattern.pipe
  import context.dispatcher

  var state = 10
  val http = Http(context.system)
  val pool = http.cachedHostConnectionPoolTls[Int](apiEndpoint.authority.host.toString())

  override def preStart() = {
    Source.single(HttpRequest(uri = "http://akka.io") -> 42)
      .via(poolClientFlow)
      .runWith(Sink.head)
      .pipeTo(self)
  }

  def receive = {
    case (res, ref) => ref match {
      case 42 => state -= 1 // Do something with the response 
    }
  }
}

如果是这样，我们为什么需要它？在文档中找不到解释如果不是，正确的模式是什么？

谢谢

【问题讨论】：

如果在你的 future 完成后你需要改变一个 actor 的状态，那么是的，pipeTo 是正确的做法，因为如果你不这样做，你可能会遇到并发 mods 到那个状态并输掉演员的利益。如果你不需要在未来完成后改变/处理状态，那么不，你不需要管道回到自我。
非常感谢，您想把它作为答案吗？
@NicholasJoseph，完成。

标签： scala akka akka-http

【解决方案1】：

正如我在评论中提到的，如果您需要改变参与者的内部状态，您应该使用pipeTo 将Future 的结果发送回您自己进行处理。如果您不这样做，您将面临并发修改该内部状态的问题的风险，并失去使用参与者的好处。 Future 的后完成逻辑不会在参与者邮箱处理的上下文中执行，因此可能与邮箱中的消息同时执行，从而导致潜在的并发问题。这就是为什么如果您需要在 Future 完成后更改状态，建议您通过管道返回给自己。

现在，如果之后没有要管理的状态，那么您不必使用pipeTo，因为并发修改状态不会成为问题。

【讨论】：