Akka HTTP 未来响应为时已晚答案

【问题标题】：Akka HTTP Future response too lateAkka HTTP 未来响应为时已晚
【发布时间】：2018-03-07 14:20:14
【问题描述】：

我想从清漆中禁止 URL。我首先要做的是从 consul 那里收集所有健康的 IP。

private def nodeInfo:Future[List[NodeInfo]] = {
  val request = HttpRequest(method = HttpMethods.GET, uri = consulUrl)

  Future.successful {
    Http().singleRequest(request).flatMap(response =>
      response.status match {
        case OK => Unmarshal(response).to[List[NodeInfo]]
        case _ =>
          response.entity.toStrict(5.seconds).flatMap { entity =>
            val body = entity.data.decodeString("UTF-8")
            log.warning(errorMessage(request, response, body))
            Future.failed(new IOException(errorMessage(response, body)))
          }
      })
  }.flatMap(value => value)
}

这按预期工作。在 for comprehension 的帮助下，我想遍历所有这些。

def banFromCache(toBanUrl:String): Future[String] = {
  for {
    nodes <- nodeInfo
    result <- loopNodes(nodes, toBanUrl)
  } yield result
}

通过 foreach 循环，我发送 HttpRequest 并为每个循环获取 HttpResponses。但是由于 Future 功能导致的结果是在请求完成之前完成的。

private def loopNodes(nodes:List[NodeInfo], toBanUrl:String):Future[String] = Future {
  val banResult = new ListBuffer[String]

  nodes.foreach(node => {
    banAllHealthy(node, toBanUrl).onComplete {
      case Failure(err) =>
        banResult += node.Node.Address + " " + err.getMessage
        log.error("Request failed: " + node.Node.Address + " " + err.getMessage)
      case Success(res) =>
        banResult += node.Node.Address + " " + res.toString
        log.info("Request success: " + node.Node.Address + " " + res.toString)
    }
  })

  banResult.toList.toString()
}

private def banAllHealthy(nodeInfo:NodeInfo, toBanUrl: String):Future[HttpResponse] = {
  def request(): Future[HttpResponse] =
    Http().singleRequest(HttpRequest(method = HttpMethods.GET, uri = "http://localhost:9000/healthcheck"))
    //Http().singleRequest(HttpRequest(method = HttpMethods.GET, uri = "http://" + nodeInfo.Node.Address + "/" + toBanUrl))

  val responseFuture: Future[HttpResponse] = request()
  responseFuture
}

这里的路线很简单：

} ~ pathPrefix("ban") {
      pathPrefix(Segment) { banpath =>
        pathEndOrSingleSlash {
          get {
            complete(banFromCache(banpath).map(_.asJson))
          }
        }
      }

有没有办法一次显示所有响应？

【问题讨论】：

标签： scala akka-http

【解决方案1】：

要累积结果字符串，请留在Future 的上下文中：

private def loopNodes(nodes: List[NodeInfo], toBanUrl: String): Future[String] = {

  val futures: List[Future[String]] = nodes.map { node =>
    banAllHealthy(node, toBanUrl)
      .map(res => s"${node.Node.Address} ${res}")
      .recover { case err => s"${node.Node.Address} ${err.getMessage}" }
  }

  Future.reduceLeft(futures)(_ + "\n" + _)
}

【讨论】：

【解决方案2】：

如果您将回调用于处理每个未来的结果，您将失去对执行的控制。使用 foreach，每个未来都存在于自己的并行执行中。父期货收益率，因为它不等待任何东西。我建议不要使用 ListBuffer 并使用更不可变的样式。无论如何，尝试将整个计算构建为封装整个计算的一个未来：

private def loopNodes(nodes:List[NodeInfo], toBanUrl:String):Future[String] = {
  nodes.map { node =>
      // Creates a tuple to store current node and http result
      // (Node, result, HttpResult)
      (node, "", banAllHealthy(node, toBanUrl))
  }.foldLeft(Future(""))((str, b) =>
      b match {
        case (node, str ,response) => {
          // Each response will be transformed to string
          (response map (result => str  + " " + node.Node.Address + " " + "Success"))
          // In case of node is not available its suppose that the HttpClient will raise an execption  
          .recover {
            case err: Throwable  =>
              str + " " + node.Node.Address + " " + "Error " + err.getMessage
            case _ =>
              str  + " " + node.Node.Address + " " + "Unknown Error"
          }
        }
    })
}

【讨论】：