等待一个超时的期货序列，而不会在TimeoutException上失败

Question

我有一系列相同类型的scala期货。

我想在一段有限的时间后，得到整个序列的结果。虽然有些期货可能已经成功，有些可能已经失败，有些还没有完成，但未完成的期货应该被视为失败。

我不想使用“依次等待每个未来”。

我确实看了这个问题：Scala等待未来序列，并尝试从那里使用解决方案，即：

  private def lift[T](futures: Seq[Future[T]])(implicit ex: ExecutionContext) =
    futures.map(_.map { Success(_) }.recover { case t => Failure(t) })

  def waitAll[T](futures: Seq[Future[T]])(implicit ex: ExecutionContext) =
    Future.sequence(lift(futures))

  futures: Seq[Future[MyObject]] = ...
  val segments = Await.result(waitAll(futures), waitTimeoutMillis millis)

但是我仍然有一个超时异常，我想是因为一些未来还没有完成。答案还指出，

现在，Future.sequence(lifted)将在每个Future完成时完成，并将使用Try表示成功和失败。

但是我希望我的Future在超时之后完成，而不是在序列中的每个Future都完成时。我还能做什么？

Answer 1

这是使用monix的解决方案

import monix.eval.Task
import monix.execution.Scheduler

val timeoutScheduler = Scheduler.singleThread("timeout") //it's safe to use single thread here because timeout tasks are very fast

def sequenceDiscardTimeouts[T](tasks: Task[T]*): Task[Seq[T]] = {
  Task
    .parSequence(
      tasks
        .map(t =>
          t.map(Success.apply) // Map to success so we can collect the value
            .timeout(500.millis)
            .executeOn(timeoutScheduler) //This is needed to run timesouts in dedicated scheduler that won't be blocked by "blocking"/io work if you have any
            .onErrorRecoverWith { ex =>
              println("timed-out")
              Task.pure(Failure(ex)) //It's assumed that any error is a timeout. It's possible to "catch" just timeout exception here
            }
        )
    )
    .map { res =>
      res.collect { case Success(r) => r }
    }
}

测试代码

implicit val mainScheduler = Scheduler.fixedPool(name = "main", poolSize = 10)


def slowTask(msg: String) = {
  Task.sleep(Random.nextLong(1000).millis) //Sleep here to emulate a slow task
    .map { _ =>
      msg
    }
}


val app = sequenceDiscardTimeouts(
  slowTask("1"),
  slowTask("2"),
  slowTask("3"),
  slowTask("4"),
  slowTask("5"),
  slowTask("6")
)

val started: Long = System.currentTimeMillis()
app.runSyncUnsafe().foreach(println)
println(s"Done in ${System.currentTimeMillis() - started} millis")

这将为每次运行打印不同的输出，但它应该如下所示

timed-out
timed-out
timed-out
3
4
5
Done in 564 millis

请注意使用了两个独立的调度程序。这是为了确保即使< code>main调度程序忙于业务逻辑，也会触发超时。您可以通过减少主调度程序的< code>poolSize来测试它。

Answer 2

如果我使用raw Future(而不是内置了这种功能的IO monad，或者没有专门用于这种功能的Akka实用程序)，我会编写如下实用程序:

// make each separate future timeout
object FutureTimeout {
  // separate EC for waiting
  private val timeoutEC: ExecutorContext = ...

  private def timeout[T](delay: Long): Future[T] = Future {
    blocking {
      Thread.sleep(delay)
    }
    throw new Exception("Timeout")
  }(timeoutEC)

  def apply[T](fut: Future[T], delat: Long)(
    implicit ec: ExecutionContext
  ): Future[T] = Future.firstCompletedOf(Seq(
    fut,
    timeout(delay)
  ))
}

然后

Future.sequence(
  futures
    .map(FutureTimeout(_, delay))
    .map(Success(_))
    .recover { case e => Failure(e) }
)

由于每个未来至多在< code>delay之后终止，我们将能够在那之后立即将它们收集到一个结果中。

但你必须记住，无论你如何触发超时，你都不能保证超时的未来会停止执行。它可以在某个地方的某个线程上运行，只是你不会等待结果。firstCompletedOf只是让这个种族更加明确。

其他一些实用程序(例如Cats Effect IO)允许您取消计算(这在像这样的比赛中使用),但是您仍然必须记住JVM不能任意“杀死”一个正在运行的线程，所以取消会发生在一个计算阶段完成之后，下一个阶段开始之前(例如在< code >之间)。映射或< code >。平面图。

如果你不害怕添加外部deps，还有其他（更可靠的，因为Thread.sleep只是一个暂时的丑陋的黑客）方法来计时未来，比如Akka utils。另请参阅类似这样的其他问题。