Akka未来-需要建议

Question

我有一个场景，其中我得到一个String消息列表，我必须遍历String并调用另一个方法，这是一个长时间运行的过程。然后我必须收集这个长时间运行过程的结果并连接结果并将其发送回用户交互界面。我对Scala中的这些未来概念很陌生。我正在使用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) }
  }
}

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

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)

请指教！

Answer 1

在我看来，对于您想要的东西，在这里只使用Scala的并行集合而不是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

Answer 2

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

Answer 3

我将在这里假设您的意思是要连接字符串。首先是对代码的一些评论：

> < li>

不需要将整个块包装在异步中，只需返回最终的未来。

值类型都可以推断，您不需要显式地声明它们

映射列表将返回列表。对结果调用 toList 是多余的。

List#sum仅适用于数字，请使用fold↓代替字符串。

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("")(_ + _)) }
  }

更新

使用Future也是一样的。折叠，正如维克多所建议的

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)) }

如果您想处理每个元素的单个错误，那么您应该看看这个答案中使用的技术。