Scala重试期货序列，直到它们全部完成答案

【问题标题】：Scala retry sequence of futures until they all completeScala重试期货序列，直到它们全部完成
【发布时间】：2020-04-25 11:38:50
【问题描述】：

在 scala 中，您将如何编写一个函数，该函数接受一个 Future 序列，将它们全部运行，不断重试任何失败并返回结果？

例如，签名可能是：

  def waitRetryAll[T](futures: Seq[Future[T]]): Future[Seq[T]]

可配置超时的奖励积分，此时函数失败并且被调用者可以处理这种情况。
如果该错误案例处理程序可以接收失败的 Futures 列表，则奖励积分。

谢谢！

【问题讨论】：

嗯...无限制地重试，退避似乎是双脚射击自己的好方法。

标签： scala future retry-logic

【解决方案1】：

基于Retry a function that returns a Future考虑

def retry[T](expr: => Future[T], n: Int = 3): Future[Either[Throwable, T]] = {
  Future
    .unit
    .flatMap(_ => expr).map(v => Right(v))
    .recoverWith {
      case _ if n > 1 => retry(expr, n - 1)
      case e => Future.failed(e).recover{case e => Left(e)}
    }
}

结合

Future.sequence

将List[Future[T]] 转换为Future[List[T]]。但是sequence 具有快速失败的行为，因此我们必须将Future[T] 提升为Future[Either[Throwable, T]]

将这些部分放在一起，我们可以定义

def waitRetryAll[T](futures: List[() => Future[T]]): Future[List[Either[Throwable, T]]] = {
  Future.sequence(futures.map(f => retry(f.apply())))
}

并像这样使用它

val futures = List(
  () => Future(42),
  () => Future(throw new RuntimeException("boom 1")),
  () => Future(11),
  () => Future(throw new RuntimeException("boom 2"))
)

waitRetryAll(futures)
  .andThen { case v => println(v) }

哪个输出

Success(List(Right(42), Left(java.lang.RuntimeException: boom 1), Right(11), Left(java.lang.RuntimeException: boom 2)))

我们可以collectLefts 或Rights 并相应地恢复或继续处理，例如

waitRetryAll(futures)
  .map(_.collect{ case v if v.isLeft => v })
  ...

请注意我们必须传递List[() => Future[T]] 而不是List[Future[T]] 以防止期货急切地启动。

【讨论】：

Either而不是Try的任何原因？
谢谢@mario-galic 我可能做错了什么，但是对于我来说这个测试用例pastebin.com/ra0TLg4a 看起来这不会重试失败的调用。该测试使用仅在第 3 次尝试后返回的函数调用 waitRetryAll。我的测试函数doItEventually（在 Future 中返回值或引发异常）对这些函数有效吗？
@Jethro 你说得对，我的实现出现了重大错误！感谢您对其进行测试。请参阅我最近尝试的编辑答案。请注意，我传递的是List[() => Future[T]] 而不是List[Future[T]]，以防止期货急于启动。在 scastie 中尝试scastie.scala-lang.org/c1agpT5PSki8PX3C0GfYKw
@ig-dev 使用Try 的解决方案也可以。我使用Either 没有特别的原因，只是为了展示一个潜在的解决方案。也许一个优势可能是 scala-cats 提供了更多与 Either 一起工作的好东西。这可能是由于，理论上，Try 不是一个单子：stackoverflow.com/questions/58103898/…

【解决方案2】：

据我记得在标准库中没有Future timeout 的实用程序。

您将如何中断/取消 JVM 上正在进行的计算？在一般情况下，您不能，您只能在wait 上中断Thread，但如果它从不waits？用于异步计算（定义取消）的 IO 库将 IO 执行为一系列较小的不可中断任务（每个 map/flatMap 创建一个新步骤），如果它们收到取消/超时，那么它们将继续执行当前任务（因为它们无法停止它) 但他们不会开始下一个。您可以在超时时返回异常，但仍然会执行最后一步，因此如果它是一些副作用（例如数据库操作），它将在您返回失败后完成。

这是不直观且棘手的，我认为这就是为什么没有将这种行为添加到标准库中的原因。

此外，未来正在进行中，可能会产生副作用。您不能获取 Future[A] 类型的值并重新运行它。但是，您可以将 future 作为 by-name 参数传递，以便在 .recoverWith 中您可以重新创建 future。

很遗憾，您可以实现类似“重试直到 LocalDateTime.now - startTime >= ”，因为我认为这是您想要的：

def retry[A](future: => Future[A], attemptsLeft: Int, timeoutTime: Instant) =
  future.recoverWith {
    case error: Throwable =>
      if (attemptsLeft <= 0 || Instant.now.isAfter(timeoutTime)) Future.failure(error)
      else retryHelper(future, attemptsLeft - 1, timeoutTime)
  }

这可以与Future.sequence 结合以创建结果列表：

def retryFutures[A](list: List[() => Future[A]) = {
  val attempts: Int = ...
  val timeout: Instant = ...
  Future.sequence(list.map(future => retry(future(), attempts, timeout)))
}

如果您想跟踪哪个未来失败和哪个成功：

def futureAttempt[A](future: Future[A]): Future[Either[Throwable, A]] =
  future.map(a => Right(a))).recover {
    case error: Throwable => Left(error)
  }

def retryFutures[A](list: List[() => Future[A]) = {
  val attempts: Int = ...
  val timeout: Instant = ...
  Future.sequence(list.map(future => retry(futureAttempt(future()), attempts, timeout)))
}

如果您不担心取消 JVM 上的期货，并且如果您有更多类似的情况，我建议您使用库。

如果您想使用为您实现重试的东西，请使用cats-retry

如果您想在定义计算时获得比 Future 更好的东西（例如，不需要您使用别名参数或空函数的东西）试试 Monix 或 ZIO (https://zio.dev/)

【讨论】：