Akka Future - 需要的建议答案

【问题标题】：Akka Future - Suggestions NeededAkka Future - 需要的建议
【发布时间】：2014-03-10 12:13:30
【问题描述】：

我有一个场景，其中我得到一个字符串消息列表，我必须遍历字符串并调用另一个方法，这是一个长时间运行的过程。然后，我必须收集这个长时间运行的过程的结果并将结果连接起来并将其发送回用户界面。我对 Scala 中的这些 Future 概念很陌生。我正在使用 Play 框架，其中字符串列表将来自用户界面。这是我第一次尝试实现 ht 场景的样子：

def futuresTest(strList: List[String]) = Action {
  Async {

    val ftrList: List[Future[String]] =
      strList.map(s => Akka.future {longRunningCall(s)} ).toList

    val futureResultList = Future.sequence(ftrList)

    val jsonResponse: String =
      futureResultList.map(_.sum).asInstanceOf[String]

    Akka.future { Ok(jsonResponse) }
  }
}

为简单起见，longRunningCall 只返回一个字符串。稍后我会将它与原始实现联系起来。

def longRunningCall(s: String) = "test"

我的问题是在声明中：

val ftrList: List[Future[String]] =
  strList.map(s => Akka.future {longRunningCall(s)} ).toList

我会假设 ftrList 将被异步填充，当它到达下一行时，我保证 futureResultList 将包含所有元素（即 strList 和 futureResultList 的大小将相等？

val futureResultList = Future.sequence(ftrList)

请指教！

【问题讨论】：

当你在List 上调用.map 时，它会返回一个List...你不需要在结果上调用.toList。
我也很惊讶地看到 .sum 处理 List[String] 和 Future 被直接转换为 String 显然没有问题。
这是一个错误。总和是我的错误。然而，IntelliJ 并没有对此有任何抱怨！
我是否正确假设 sum 应该在这里连接字符串？

标签： scala playframework akka future

【解决方案1】：

我在这里假设您的意思是要连接字符串。先上一些cmets的代码：

无需将整个块包装在 Async 中，只需将您返回的最终未来包装。
值类型均可推断，无需显式声明
映射List 会返回List。对结果调用toList 是多余的。
List#sum 仅适用于数字，使用 foldLeft 代替字符串。
Future[String] 可以不通过asInstanceOf 直接转换为String。无论如何，您都会以Future 的形式返回它。
map 和 Future.sequence 可以合并为一个 Future.traverse 操作。

并更改代码以应用这些要点：

def futuresTest(strList: List[String]) = Action {

  val ftrList = Future.traverse(strList) {
    s => Future( longRunningCall(s) )
  }

  // this will be a Future[String]
  val jsonResponse = ftrList map { _.foldLeft("")(_ + _) }

  Async { jsonResponse map (OK(_)) }
}

最后两行也可以合并：

  Async {
    ftrList map { xs => OK(xs.foldLeft("")(_ + _)) }
  }

更新

正如 Viktor 所建议的那样，使用 Future.fold 也是如此

def futuresTest(strList: List[String]) = Action {     
  val ftrList = strList map { longRunningCall(_) } // List[Future]
  val jsonResponse = Future.fold(ftrList)("")(_ + _) // Future[String]
  Async { jsonResponse map (OK(_)) }
}

为了处理失败，你想恢复未来并让它返回不同的响应：

  Async { jsonResponse map (OK(_)) recover (InternalServerError(_.toString)) }

如果您想处理每个元素的个别错误，那么您应该查看this answer 中使用的技术。

【讨论】：

Async { Ok(jsonResponse) } - 这不会编译，因为我必须将此 Future[String] 转换为结果！
啊对，它必须是Future[Response]，而不是包含未来的响应。将更新答案。
另一个问题：ftrList 是否会包含 strList 中所有元素的所有响应？我应该考虑添加一个 onComplete 处理程序吗？有什么建议吗？
ftrList 是一个单一的Future，当所有嵌套的期货都完成后，它将以List 完成。您不需要添加自己的 onComplete 处理程序，因为 Play Async 在完成时已经处理了这个未来的处理。
非常感谢您的帮助。

【解决方案2】：

只要 longRunningCall 能够正确返回值，您所做的事情就会正常工作。 ftrList 大小将等于 strList 大小，除非 longRunningCall 异常

【讨论】：

长时间运行的调用可能会引发异常。在这种情况下我该怎么办？我应该大小写匹配吗？
Akka.future {longRunningCall(s)}.recover{ case e:Exception =>{e.getLocalizedMessage()} } 这将返回错误消息而不是结果字符串

【解决方案3】：

在我看来，对于您想要的，这里只使用 Scala 的 Parallel 集合而不是 Futures 会简单得多。通过使用.par.map，您可以对列表中的每个项目并行执行长时间运行的操作并将结果收集在一起。

def getResponse(strList: List[String]) = Action {

    val results: List[String] = strList.par.map(longRunningCall(_))
    val jsonResponse: String = results.mkString(",") //concatenate using ','

    Ok(jsonResponse)
}

http://docs.scala-lang.org/overviews/parallel-collections/configuration.html

【讨论】：