1

我试图找出按顺序执行一系列期货的最巧妙方法，其中一个期货的执行取决于前一个。我正在尝试为任意数量的期货执行此操作。

用户案例：

  • 我已经从我的数据库中检索了一些 Id。
  • 我现在需要在 Web 服务上检索一些相关数据。
  • 找到有效结果后，我想停止。
  • 我只关心成功的结果。

并行执行所有这些，然后解析返回的结果集合不是一种选择。我必须一次执行一个请求，并且只有在前一个请求没有返回结果时才执行下一个请求。

当前的解决方案是沿着这些思路。使用 foldLeft 执行请求，然后仅在前一个未来满足某些条件时才评估下一个未来。

def dblFuture(i: Int) = { i * 2 }
val list = List(1,2,3,4,5)
val future = list.foldLeft(Future(0)) {
  (previousFuture, next) => {
    for {
      previousResult <- previousFuture
      nextFuture <- { if (previousResult <= 4) dblFuture(next) else previousFuture }
    } yield (nextFuture)
  }
}

这样做的最大缺点是a）即使我得到了我满意的结果，我也会继续处理所有项目；b）一旦我找到了我想要的结果，我就会继续评估谓词。在这种情况下，它是一个简单的 if，但实际上它可能更复杂。

我觉得我错过了一个更优雅的解决方案。

4

2 回答 2

5

查看您的示例，似乎先前的结果与后续结果无关，而唯一重要的是先前的结果满足某些条件以防止计算下一个结果。如果是这种情况，这里是使用filterand的递归解决方案recoverWith。

def untilFirstSuccess[A, B](f: A => Future[B])(condition: B => Boolean)(list: List[A]): Future[B] = {
    list match {
        case head :: tail => f(head).filter(condition).recoverWith { case _: Throwable => untilFirstSuccess(f)(condition)(tail) }
        case Nil => Future.failed(new Exception("All failed.."))
    }
 }

filter只有在Future完成时才会调用，并且recoverWith只有在Future失败时才会调用。

def dblFuture(i: Int): Future[Int] = Future { 
     println("Executing.. " + i)
     i * 2 
 }

val list = List(1, 2, 3, 4, 5)

scala> untilFirstSuccess(dblFuture)(_ > 6)(list)
Executing.. 1
Executing.. 2
Executing.. 3
Executing.. 4
res1: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@514f4e98

scala> res1.value
res2: Option[scala.util.Try[Int]] = Some(Success(8))
于 2014-10-29T03:42:16.607 回答
2

最简洁的方式，“真正的函数式编程”是 scalaz-stream ;) 但是，您需要从 scala Future 切换到 scalaz.concurrent.Task 以获得“未来结果”的抽象。这有点不同。Task 是纯粹的，Future 是“运行计算”，但它们有很多共同点。

  import scalaz.concurrent.Task
  import scalaz.stream.Process

  def dblTask(i: Int) = Task {
    println(s"Executing task $i")
    i * 2
  }

  val list = Seq(1,2,3,4,5)

  val p: Process[Task, Int] = Process.emitAll(list)

  val result: Task[Option[Int]] =
    p.flatMap(i => Process.eval(dblTask(i))).takeWhile(_ < 10).runLast

  println(s"result = ${result.run}")

结果：

Executing task 1
Executing task 2
Executing task 3
Executing task 4
Executing task 5
result = Some(8)

如果您的计算已经是 scala Future，您可以将其转换为 Task

implicit class Transformer[+T](fut: => SFuture[T]) {
  def toTask(implicit ec: scala.concurrent.ExecutionContext): Task[T] = {
    import scala.util.{Failure, Success}
    import scalaz.syntax.either._
    Task.async {
      register =>
        fut.onComplete {
          case Success(v) => register(v.right)
          case Failure(ex) => register(ex.left)
        }
    }
  }
}
于 2014-10-29T04:30:51.627 回答