Scala - 按顺序依次执行任意数量的Futures

Question

我正在试图找出按序列执行一系列期货的最佳方式,其中一个Future的执行取决于之前的.我试图为任意数量的期货做这件事.

用户案例:

• 我从我的数据库中检索了许多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,但实际上它可能更复杂.

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

Answer 1

看看你的例子,似乎前面的结果与后续结果没有关系,相反,唯一重要的是前一个结果满足某些条件以防止计算下一个结果.如果是这种情况,这里是一个使用filter和的递归解决方案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))