喷雾http客户端和数千个请求

Question

akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka://smallTasks/user/IO-HTTP#151444590]] after [15000 ms]
akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka://smallTasks/user/IO-HTTP#151444590]] after [15000 ms]
akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka://smallTasks/user/IO-HTTP#151444590]] after [15000 ms]
akka.pattern.AskTimeoutException: Ask timed out on 

  implicit val system = ActorSystem("smallTasks")
  implicit val timeout = new Timeout(15.seconds)

  import system.dispatcher

  def doHttpRequest(url: String): Future[HttpResponse] = {
    (IO(Http) ? HttpRequest(GET, Uri(url))).mapTo[HttpResponse]
  }

def getOnlineOffers(modelId: Int, count: Int = 0): Future[Any] = {

    val result = Promise[Any]()

    AkkaSys.doHttpRequest(Market.modelOffersUrl(modelId)).map(response => {
      val responseCode = response.status.intValue
      if (List(400, 404).contains(responseCode)) {
        result.success("Bad request")
      } else if (responseCode == 200) {
        Try {
          Json.parse(response.entity.asString).asOpt[JsObject]
        } match {
          case Success(Some(obj)) =>
            Try {
              (obj \\ "onlineOffers").head.as[Int]
            } match {
              case Success(offers) => result.success(offers)
              case _ => result.success("Can't find property")
            }

          case _ => result.success("Wrong body")
        }
      } else {
        result.success("Unexpected error")
      }
    }).recover { case err =>
      if (count > 5) {
        result.success("Too many tries")
      } else {
        println(err.toString)
        Thread.sleep(200)
        getOnlineOffers(modelId, count + 1).map(r => result.success(r))
      }
    }

    result.future

  }

如何恰当地做到这一点？也许我需要配置akka调度程序？

Answer 1

您可以使用http://spray.io/documentation/1.2.2/spray-client/并为您编写个人管道

val pipeline: Future[SendReceive] =
      for (
        Http.HostConnectorInfo(connector, _) <-
          IO(Http) ? Http.HostConnectorSetup("www.spray.io", port = 80)
      ) yield sendReceive(connector)

    val request = Get("/segment1/segment2/...")
    val responseFuture: Future[HttpResponse] = pipeline.flatMap(_(request))

获取HttpResponse

import scala.concurrent.Await
import scala.concurrent.duration._
val response: HttpResponse = Aweit(responseFuture, ...)

转换

import spray.json._
response.entity.asString.parseJson.convertTo[T]

Try(response.entity.asString.parseJson).isSuccess