Akka HTTP 连接池在几个小时后挂起

我有一个 HTTP 连接池，在运行几个小时后挂起：

private def createHttpPool(host: String): SourceQueue[(HttpRequest, Promise[HttpResponse])] = {
    val pool = Http().cachedHostConnectionPoolHttps[Promise[HttpResponse]](host)
    Source.queue[(HttpRequest, Promise[HttpResponse])](config.poolBuffer, OverflowStrategy.dropNew)
      .via(pool).toMat(Sink.foreach {
        case ((Success(res), p)) => p.success(res)
        case ((Failure(e), p)) => p.failure(e)
      })(Keep.left).run
  }

我将项目排队：

private def enqueue(uri: Uri): Future[HttpResponse] = {
    val promise = Promise[HttpResponse]
    val request = HttpRequest(uri = uri) -> promise

    queue.offer(request).flatMap {
      case Enqueued => promise.future
      case _ => Future.failed(ConnectionPoolDroppedRequest)
    }
}

并像这样解决响应：

private def request(uri: Uri): Future[HttpResponse] = {
    def retry = {
      Thread.sleep(config.dispatcherRetryInterval)
      logger.info(s"retrying")
      request(uri)
    }

    logger.info("req-start")
    for {
      response <- enqueue(uri)

      _ = logger.info("req-end")

      finalResponse <- response.status match {
        case TooManyRequests => retry
        case OK => Future.successful(response)
        case _ => response.entity.toStrict(10.seconds).map(s => throw Error(s.toString, uri.toString))
      }
    } yield finalResponse
}

如果 Future 成功，则该函数的结果总是会被转换：

def get(uri: Uri): Future[Try[JValue]] = {
  for {
    response <- request(uri)
    json <- Unmarshal(response.entity).to[Try[JValue]]
  } yield json
}

一切工作正常一段时间，然后我在日志中看到的只是 req-start 而没有 req-end。

我的akka​​配置是这样的：

akka {
  actor.deployment.default {
    dispatcher = "my-dispatcher"
  }
}

my-dispatcher {
  type = Dispatcher
  executor = "fork-join-executor"

  fork-join-executor {
    parallelism-min = 256
    parallelism-factor = 128.0
    parallelism-max = 1024
  }
}

akka.http {
  host-connection-pool {
    max-connections = 512
    max-retries = 5
    max-open-requests = 16384
    pipelining-limit = 1
  }
}

我不确定这是配置问题还是代码问题。我的并行度和连接数如此之高，因为没有它，我的请求/秒速率非常差（我想尽可能快地请求 - 我有其他速率限制代码来保护服务器）。

您没有消耗从服务器返回的响应实体。引用以下文档：

消耗（或丢弃）请求的实体是强制性的！如果 意外地留下既没有消耗也没有丢弃 Akka HTTP 将假设 传入的数据应保持背压，并将停止 通过 TCP 反压机制传入数据。客户应该 无论 HttpResponse 的状态如何，都会使用实体。

该实体以以下形式出现：Source[ByteString, _]需要运行它以避免资源匮乏。

如果您不需要读取实体，则使用实体字节的最简单方法是丢弃它们，方法是使用

res.discardEntityBytes()

（您可以通过添加 - 例如 - 来附加回调.future().map(...)).

文档中的此页面 http://doc.akka.io/docs/akka-http/10.0.0/scala/http/implications-of-streaming-http-entity.html描述了所有替代方案，包括如何在需要时读取字节。

--- EDIT

提供更多代码/信息后，很明显资源消耗不是问题。此实现中还有另一个大危险信号，即Thread.sleep在重试方法中。 这是一个阻塞调用，很可能会导致底层 Actor 系统的线程基础设施处于饥饿状态。

详细解释了为什么这是危险的docs http://doc.akka.io/docs/akka-http/10.0.3/scala/http/handling-blocking-operations-in-akka-http-routes.html.

尝试改变它并使用akka.pattern.after (docs http://doc.akka.io/docs/akka/current/scala/futures.html#After）。下面的例子：

def retry = akka.pattern.after(200 millis, using = system.scheduler)(request(uri))