如何在多个未来等待中检索部分结果，当某些可能超时？

Question

我在scala on play框架中使用FUTURE。但我很难在合并多个期货时出现超时的情况下获得部分最终结果。下面是我的代码的功能。它有两个未来分别查询两个提供者。然后使用for/yield语句合并结果。然后等待具有超时值的结果。当两个提供者按时回复查询时，它可以正常工作。但是，如果只有一个提供者按时回复，我知道等待将超时，但在这种情况下，我仍然需要检索从另一个按时回复的提供者返回的数据。我该怎么做？

val pool = Executors.newCachedThreadPool()
implicit val ec = ExecutionContext.fromExecutorService(pool)

var future1 = Future(QueryProvider(provider1, request1))  
var future2 = Future(QueryProvider(provider2, request2))  

val future = for {
    result1 <- future1
    result2 <- future2
 } yield (Merge(result1, result2))

 val duration = Duration(60000, MILLISECONDS)

 try{
     result = Await.result(future, duration).asInstanceOf[string]       
 }

catch{
    case _: Exception => println("time out...") 
    //Here how can I retrieve provider1's result if only provider2 timeout???***
 }

Answer 1

您可以在akka中的之后使用而不是阻塞Await.result：

val timeout =
  akka.pattern.after(FiniteDuration(60000, MILLISECONDS), using = system.scheduler){
    Future.successful(None)
  }

val resFuture = for {
    result1 <- Future firstCompletedOf Seq(future1.map{Some(_)}, timeout)
    result2 <- Future firstCompletedOf Seq(future2.map{Some(_)}, timeout)
} yield (result1, result2)

val result = resFuture map {
  case (Some(r1), Some(r2)) => Merge(r1, r2)
  case (Some(r1), None) => PartialResult1(r1)
  case (None, Some(r2)) => PartialResult2(r2)
  case _ => EmptyResult
}

在这种情况下，resFuture将在60秒内完成，您可以处理部分结果。此外，您不需要在播放中等待，也可以使用异步。

如果你有许多相同类型的等价期货，你可以这样使用它：

val futures: Seq[Future[Int]] = ???
val futureWithTimeout =
  futures.map{ f => Future firstCompletedOf Seq(f.map{Some(_)}, timeout) }

val result: Future[Seq[Option[Int]]] = Future.sequence(futureWithTimeout)

// In case you need to know index of completed or not completed future
val indexedResults = result.zipWithIndex

// In case you just need only completed results
val completedResults: Future[Seq[Option[Int]]] = result.map{_.flatten}

这里的类型只是为了说明。