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

Question

在 scala 中，您将如何编写一个函数来获取期货序列，运行所有函数，不断重试任何失败的函数，并返回结果？

例如，签名可以是：

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

加分为一个可配置的超时，在这一点上的功能失败，被调用者可以处理这种情况。< br >如果错误案例处理程序可以接收失败的期货列表，则奖励加分。

谢谢！

Answer 1

匿名用户

据我所知，在标准库中没有用于< code>Future超时的实用程序。

您将如何中断/取消JVM上正在进行的计算？在一般情况下，您不能，您只能在线程处于等待状态时中断它，但如果它永远不会等待。用于异步计算（定义取消）的IO库将IO作为一系列较小的不可中断任务执行（每个map/Map平坦创建一个新步骤），如果它们收到取消/超时，它们将继续执行当前任务（因为它们无法停止它），但它们不会开始下一个任务。您可以在超时时返回异常，但仍然会执行最后一步，因此如果它是一些副作用（例如：DB操作）它将在您已经返回失败后完成。

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

此外，未来是正在进行的，潜在的副作用操作。您不能接受< code>Future[A]类型的值并重新运行它。但是，您可以通过名称参数传递future，这样在< code >中。recoverWith您可以重新创造未来。

可悲的是，你可以实现类似“重试到LocalDateTime.now开始时间”的东西

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

它可以与< code>Future.sequence结合来创建结果列表:

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

如果要跟踪哪个future失败，哪个future成功：

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

如果你想在定义计算方面有比< code>Future更好的东西(例如，不需要你使用按名参数或空函数的东西),试试Monix或ZIO(https://zio.dev/)

Answer 2

基于Retry返回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

它将列表[Future[T]]转换为未来[List[T]]。然而，序列具有快速故障行为，因此我们必须将未来[T]，提升到将来[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)))

我们可以收集我们的左s或右s并相应地恢复或继续处理，例如

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

注意我们如何传入List[（）=