未来递归模式/任意长度的未来链

Question

我很好奇递归构建Akka期货链的最佳方式，它将按顺序运行，如果未来的< code>doWork调用失败，未来将重试3次，如果重试次数用完，链将失败。假设所有< code>doWork调用都通过，则返回的futChain应该只完成。

object Main extends App {
  val futChain = recurse(2)

  def recurse(param: Int, retries: Int = 3): Future[String] {
    Future {
      doWorkThatMayFailReturningString(param...)
    } recoverWith {
      case e => 
        if (retries > 0) recurse(param, retries -1)
        else  Future.failed(e)
    } flatMap {
      strRes => recurse(nextParam) //how should the res from the previous fut be passed?
    }
  }

  futChain onComplete {
    case res => println(res) //should print all the strings
  }
}

< li >如何将结果作为集合获取？即，在本例中，从< code>doWork函数返回的每个< code>String(我需要以某种方式修改< code>recurse函数，以返回一个< code > future[List[String]] < li >我应该使用< code>recover还是< code>recoverWith？ < li >可以调用< code>flatMap来链接这些调用吗 < li >我应该考虑尾部递归吗

Answer 1

您可以像这样实现可重试的Future：

def retry[T](f: => Future[T])(n: Int)(implicit e: ExecutionContext): Future[T] = {
    n match {
        case i if (i > 1) => f.recoverWith{ case t: Throwable => retry(f)(n - 1)}
        case _ => f
    }       
}

这没有针对尾部html" target="_blank">递归进行优化，但是如果您只打算重试几次，则不会得到堆栈溢出（我想如果它在前几个失败，无论如何它都会继续失败）。

然后我会单独做链接。如果您有有限数量的函数要链接在一起，每个函数都依赖于前面的函数(出于某种原因，您希望聚合结果)，您可以使用< code > for comprehensions(< code > flat map 的语法糖):

for {
    firstResult <- retry(Future(doWork(param)))(3)
    secondResult <- retry(Future(doWork(firstResult)))(3)
    thirdResult <- retry(Future(doWork(secondResult)))(3)
} yield List(firstResult, secondResult, thirdResult)

对于任意长的链，您可以使用 Future.sequence（Akka 库中的 Futures）并行执行它们：

def doWork(param: String): String = ...

val parameters: List[String] = List(...)

val results: Future[List[String]] = Future.sequence(parameters.map(doWork(_)))

这将解开本来是< code > List[Future[String]]到< code > Future[List[String]]的关系。

以下是按顺序执行类似操作的一种方法：

def sequential[A, B](seq: List[A])(f: A => Future[B])(implicit e: ExecutionContext): Future[List[B]] = {
    seq.foldLeft(Future.successful(List[B]())) { case (left, next) =>
        left.flatMap(list => f(next).map(_ :: list))
    }
}

def doWork(param: String): String = ...

val results: Future[List[String]] = sequential(parameters)(param => Future(doWork(param))) 

这些功能的实现对您的用例非常敏感。如果链中的任何期货失败，上述两个函数将返回失败的期货。有时候你会想要这个，有时候不会。如果您想只收集成功的期货，并放弃失败的期货而不放弃整个结果，您可以添加一个额外的步骤来恢复失败。

此外，recover 和 recoverWith 之间的区别在于它接受的 PartialFunction 的类型。recover 将失败的 future 替换为默认值，而 recoverWith 使用另一个 Future 执行此操作。在我重试的情况下，recoverWith 更合适，因为我正在尝试用自身来恢复失败的未来。