Scala：地图值中方法的重复执行

Question

我已经实现了一个 Scala 函数，它使用来自 Apache HttpComponents 的 HttpClient v4.5.3 执行 HTTP 请求。这个想法是解析响应并检查它是否正常（即返回代码 200）。如果不是，请重试 n 次。

当响应正常时，一切都按预期进行。 我可以从记录器输出中看到，在最初出现 HTTP 错误的情况下，重试也会按预期执行。但是，当它失败了n次时，它似乎又重新开始了整个过程。

这是执行请求的函数：

import org.apache.http.client.methods.{CloseableHttpResponse, HttpPost, HttpUriRequest}
[...]

class SearchQuery {
  [...]
  def executeRequest(retries: Int = 2, delay: Int = 1): Option[SearchResponse] = {
      val request: HttpUriRequest = makeRequest
      val response: CloseableHttpResponse = config.httpClient.execute(request)
      val searchResults: Option[SearchResponse] = parseResponse(response)
      response.close()
      if (searchResults.isEmpty && retries > 0) {
        logger.warn(s"Failed to retrieve response from ${source}. " +
          s"Retrying $retries more time(s) in $delay second(s)...")
        Thread.sleep(delay * 1000)
        executeRequest(retries - 1, delay)
      }
      else searchResults
  }
  [...]
}

makeRequest() 方法生成 HTTP 请求。 config.httpClient 提供可重复使用的 HttpClient 实例。

除了递归中的调用外，executeRequests() 的调用如下所示。这个想法是有一个或多个服务器（源），由调用者定义，调用是并行执行的；它们在enum 类型Source 中定义。 queryFromfile() 读取给定文件的内容以生成 SearchQuery 对象，然后从中调用 executeRequest。

val sources: Set[Source] = Set(Source.1)
val file: File = ...
val responses: parallel.ParMap[Source, Option[SearchResponse]] = sources
  .par
  .map(source => (source, SearchQuery.queryFromFile(file, source)))
  .toMap
  .mapValues(query => query.executeRequest())

在本例中，executeRequest 应该为每个源调用一次。一个来源顺利通过，另一个预计在这里失败。

来自日志：

14:56:48.656 [scala-execution-context-global-15] WARN  query.SearchQuery - Failed to retrieve response from source1. Retrying 2 more time(s) in 3 second(s)...
14:57:07.136 [scala-execution-context-global-15] WARN  query.SearchQuery - Failed to retrieve response from source1. Retrying 1 more time(s) in 3 second(s)...
14:57:25.538 [ScalaTest-run-running-FileProcessorTest] ERROR process.FileProcessor - Failed to retrieve results for file 'XXX' from sources: source1. Continuing.
14:57:40.933 [scala-execution-context-global-15] WARN  query.SearchQuery - Failed to retrieve response from source1. Retrying 2 more time(s) in 3 second(s)...
14:57:59.214 [scala-execution-context-global-15] WARN  query.SearchQuery - Failed to retrieve response from source1. Retrying 1 more time(s) in 3 second(s)...

使用 IntelliJ 调试器，逻辑看起来也与最初的预期一样，在 n 次尝试后以 else searchResults 行结束。然而，执行程序并没有实际返回结果，而是跳回到if 子句，进入递归调用executeRequest(retries - 1, delay)。

更新： 我发现这种行为是由responses val 上的后续操作引起的：

val emptyResponses = responses.filter(_._2.isEmpty)
if (emptyResponses.nonEmpty)
  logger.error(
    s"Failed to retrieve results for '$fileName' from sources: " +
      s"${emptyResponses.keys.mkString(",")}. Continuing."
  )

在这里，我检查是否有任何调用失败并记录错误。我不明白为什么这会触发executeRequest() 第二次调用。为什么会这样？

此外，即使在概念上看起来非常相似，以下行也不会导致第二次调用 executeRequest()：

responses.values
  .filter(response => response.isDefined)
  .map(response => response.get)

Answer 1

executeResponse() 在访问值时再次执行的原因是每次访问时都会重新计算 Scala Map 的值。 responses 因此需要建模为元组(source, Option[SearchResponse]：

val responses = sources
  .par
  .map(source => (source, SearchQuery.queryFromFile(file, source)))
  .map(pair => (pair._1, pair._2.executeRequest()))

这基本上消除了mapValues 的简洁性，但导致元组中第二个元素的值固定。