akka http文档在高级客户端请求API文档中提到,我们不应该在参与者的未来中使用access状态形式。
相反,这应该是使用的模式:
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
}
}
}如果是的话,我们为什么要这样做呢?无法在文档中找到解释,那么正确的模式是什么?
谢谢
发布于 2016-02-04 12:56:45
正如我的注释中提到的,如果需要更改参与者的内部状态,则应该使用pipeTo将Future的结果发送回自己进行处理。如果不这样做,就有可能遇到并发修改内部状态的问题,并失去使用参与者的好处。Future的后完成逻辑不会在参与者的邮箱处理上下文中执行,因此可能与邮箱中的一条消息同时执行,从而导致潜在的并发问题。这就是为什么如果在Future完成后需要更改状态,则建议回送回self。
现在,如果之后没有状态给管理器,那么您就不必使用pipeTo,因为状态的并发修改不会成为一个问题。
https://stackoverflow.com/questions/35167502
复制相似问题