通过连接池发出http请求时,Akka Flow会挂起

Question

我正在使用Akka 2.4.4并尝试从Apache HttpAsyncClient迁移(失败).

下面是我在项目中使用的代码的简化版本.

问题是,如果我向流发送超过1-3个请求,它就会挂起.经过6个小时的调试后,我甚至找不到问题.我没有看到异常,错误日志,事件Decider.没有 :)

我尝试将connection-timeout设置减少到1s,以为它可能正在等待来自服务器的响应,但它没有帮助.

我究竟做错了什么 ？

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.headers.Referer
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.http.scaladsl.settings.ConnectionPoolSettings
import akka.stream.Supervision.Decider
import akka.stream.scaladsl.{Sink, Source}
import akka.stream.{ActorAttributes, Supervision}
import com.typesafe.config.ConfigFactory

import scala.collection.immutable.{Seq => imSeq}
import scala.concurrent.{Await, Future}
import scala.concurrent.duration.Duration
import scala.util.Try

object Main {

  implicit val system = ActorSystem("root")
  implicit val executor = system.dispatcher
  val config = ConfigFactory.load()

  private val baseDomain = "www.google.com"
  private val poolClientFlow = Http()(system).cachedHostConnectionPool[Any](baseDomain, 80, ConnectionPoolSettings(config))

  private val decider: Decider = {
    case ex =>
      ex.printStackTrace()
      Supervision.Stop
  }

  private def sendMultipleRequests[T](items: Seq[(HttpRequest, T)]): Future[Seq[(Try[HttpResponse], T)]] =

    Source.fromIterator(() => items.toIterator)
      .via(poolClientFlow)
      .log("Logger")(log = myAdapter)
      .recoverWith {
        case ex =>
          println(ex)
          null
      }
      .withAttributes(ActorAttributes.supervisionStrategy(decider))
      .runWith(Sink.seq)
      .map { v =>
        println(s"Got ${v.length} responses in Flow")
        v.asInstanceOf[Seq[(Try[HttpResponse], T)]]
      }

  def main(args: Array[String]) {

    val headers = imSeq(Referer("https://www.google.com/"))
    val reqPair = HttpRequest(uri = "/intl/en/policies/privacy").withHeaders(headers) -> "some req ID"
    val requests = List.fill(10)(reqPair)
    val qwe = sendMultipleRequests(requests).map { case responses =>
      println(s"Got ${responses.length} responses")

      system.terminate()
    }

    Await.ready(system.whenTerminated, Duration.Inf)
  }
}

还有什么代理支持？似乎对我也没有用.

Answer 1

您需要完全使用响应的主体,以便连接可用于后续请求.如果您根本不关心响应实体,那么您可以将它排放到a Sink.ignore,如下所示:

resp.entity.dataBytes.runWith(Sink.ignore)

通过默认配置,当使用主机连接池时,最大连接数设置为4.每个池都有自己的队列,其中请求等待其中一个打开的连接可用.如果该队列超过32(默认配置,可以更改,必须是2的幂),那么你将开始看到失败.在您的情况下,您只执行10个请求,因此您没有达到该限制.但是,如果不消耗响应实体,则不会释放连接,其他所有内容都会排在后面,等待连接释放.