使用并行集合的scala http请求答案

【问题标题】：scala http requests using parallel collections使用并行集合的scala http请求
【发布时间】：2020-07-01 14:13:59
【问题描述】：

我正在尝试使用 scala 并行集合。我正在尝试从本地服务器获取数据，我已经设置好了，这是我的代码

val httpRequestInputs = List(inputs).par

def getResponse(data: String, url: String) = {
    val request = basicRequest.body(text).post(url)
      .headers(Map("content-type" -> "text/plain", "charset" -> "utf-8"))
      .response(asString)

    implicit val backend
    = HttpURLConnectionBackend(options = SttpBackendOptions.connectionTimeout(5.minutes))
    request.readTimeout(5.minutes).send().body
}



// program executes from here
  httpRequestInputs.foreach { input =>
      val result = getResponse(input, url)
      result match {
          case Right(value) => println(value)
          case Left(value) => println("error")
     }

当使用小尺寸的输入时，没有问题，但是，当我尝试使用大尺寸的输入时，程序抛出SocketException，我检查了服务器，服务器没有错误，在我看来，客户端正在提前关闭连接。而且，这些大投入，单独运行时，通常需要不到 90 秒的时间来获得响应。

我尝试在 http 请求中扩展连接和读取超时选项，但我仍然得到例外。

谁能帮我理解，为什么客户端要关闭连接？

对于http请求，我使用的是客户端com.softwaremill.sttp.client

【问题讨论】：

你能从抛出的SocketException 中打印出message 吗？这可能会帮助您缩小问题的范围

标签： scala http parallel-collections

【解决方案1】：

如果“大输入大小”意味着至少有几千个输入，并且每个输入都连接到同一个远程服务器，那么很可能您正在用尽运行它的临时端口范围：本质上是有一个限制（因操作系统而异）在一定时间内（Windows documentation, but every OS to my knowledge has similar limits）可以与同一远程主机和端口建立的连接数。

您要么需要捕获异常并重试，要么限制连接尝试，以免耗尽范围。（在极少数情况下，如果您尝试的次数不超过限制，则可能有一个操作系统配置选项可让您增加限制）。

使用 Scala 标准库来限制这种情况的一个好方法是使用Future：

import scala.concurrent.{ ExecutionContext, Future }

implicit val executionContext = ExecutionContext.fromExecutor(
  new java.util.concurrent.ForkJoinPool(1000) // Allow 1000 requests to be executing at once
)

val allRequestsFut =
  Future.sequence(
    httpRequestInputs.map { input =>
      Future { getResponse(input, url) }.map {
        _ match {
          case Right(value) => println(value)
          case Left(err) => println(s"error: $err")
        }
      }
    )

allRequestsFut.foreach { _ =>
  println("all requests complete")
}

请注意，许多操作系统（包括 Linux）会在端口关闭后继续保留临时端口一段时间。要动态限制请求，我建议使用 Akka Streams 之类的东西。

【讨论】：