斯卡拉如何等待几个期货

Question

我SEQ

我想做的迭代自动

Answer 1

import scala.concurrent.{Await, Future} 
import scala.concurrent.duration._ 
import scala.concurrent.ExecutionContext.Implicits.global 

def f(x:String) = Future { x } 

val functions = Seq(
() => f("a"), 
() => f("b"), 
() => f(""), 
() => f("d"), 
() => f("e")) 

functions.takeWhile(x => Await.result(x(), 5 seconds) != "") 

这会给你

res0: Seq[() => scala.concurrent.Future[String]] = List(<function0>, <function0>) 

荫不知道是否有另一种方法但为了我的理解，如果你想根据前面的未来的结果来决定，你需要等待。

编辑： 这是一个递归方法。但在我看来，与迭代器的人是好的。

import scala.concurrent.{Await, Future} 
import scala.concurrent.duration._ 
import scala.concurrent.ExecutionContext.Implicits.global 

def f(x:String) = Future { x } 

val functions = Seq(
() => f("a"), 
() => f("b"), 
() => f(""), 
() => f("d"), 
() => f("e")) 


check(functions, 0) 
def check(list:Seq[() => Future[String]], pos:Int):String = { 
    if(list.size <= pos) return "" 

    val res = Await.result(list(pos)(), 5 seconds) 
    if(condition(res)) { 
    res 
    } 
    else check(list, pos+1) 
} 

def condition(c:String):Boolean = if(c == "d") true else false 

它的尾递归，因此你的堆栈不会溢出那么容易。

Answer 2

您可以使用iterator和flatMap。下面是一个简化的例子：

import scala.concurrent.Await 
import scala.concurrent.Future 
import scala.concurrent.duration._ 
import scala.concurrent.ExecutionContext.Implicits.global 

val functions: Seq[() => Future[Int]] = Seq(
() => Future(1), 
() => Future(2), 
() => Future(3), 
() => Future(4), 
() => Future(5) 
) 

def checkResult(result: Int): Boolean = result > 3 

functions.iterator.flatMap { func => 
    val result = Await.result(func(), 5.seconds) 
    if (checkResult(result)) Some(result) else None 
}.next() 

该代码将返回。

更新：

如果你只关心正确的结果并不在意所有期货是否得到执行，你可以用fallBackTo把它们连：

import scala.util._ 

functions.tail.foldLeft[Future[Int]](functions.head().filter(checkResult)) { 
    case (result, function) => 
     result.fallbackTo(function().filter(checkResult)) 
}.onComplete { 
    case Success(result) => s"Got one: $result" 
    case Failure(exception) => s"Damn! Got NO result due to: $exception" 
} 

这将使你0123¾将与Success(4)完成与我的例子谓词x > 3。

更新2

当我发现自己有一些额外的时间，我冒昧地在其他的答案（尤其是Alexander Azarov）的一些意见和建议以贷款模式相结合方法：

import scala.annotation.tailrec 
import scala.concurrent.Future 
import scala.concurrent.ExecutionContext.Implicits.global 

val functions: Seq[() => Future[Int]] = Seq(
() => Future(1), 
() => Future(2), 
() => Future(3), 
() => Future(4), 
() => Future(5) 
) 

def checkResult(result: Int): Boolean = result > 5 

def withFirstValidResult[A, B](futureFuncs: Seq[() => Future[A]], predicate: A => Boolean, defaultValue: A)(op: (A) => B): Future[B] = { 
    @tailrec 
    @inline def find(remaining: Iterator[Future[A]]): Future[A] = { 
    if (remaining.hasNext) { 
     remaining.next().filter(predicate).recoverWith { 
     case _ => 
     find(remaining) 
     } 
    } else { 
     Future.successful { 
     println(s"No valid result found, falling back to default: $defaultValue") 
     defaultValue 
     } 
    } 
    } 


    find(futureFuncs.iterator.map(_())).map(op) 
} 

你现在可以使用这个函数来处理你想要的任何类型的期货，并提供一个应该用第一个有效结果或者defaultValue执行的操作（虽然恕我直言，如果没有有效的结果我宁愿一个Failure适当的错误处理我自己）：

withFirstValidResult(functions.iterator, checkResult, 0) { result => 
    println(s"Got a valid result: $result") 
}.onFailure { 
    case ex => 
    println(s"Got NO valid result: $ex") 
} 

作为一个额外的小费，如果你想加速功能，可以迭代懒洋洋地在与futureFuncs.iterator.buffered一次评估两种功能。这样，当一个未来的结果正在被评估时，迭代器将自动开始下一个未来。

Answer 3

def firstF[A](seq: Seq[() => Future[A]], predicate: A => Boolean): Future[A] = 
    seq.head().filter(predicate).recoverWith { 
    case _ => firstF[A](seq.tail, predicate) 
    } 

方法firstF将返回符合指定条件的第一Future。如果输入Future均不匹配，则结果Future将为Failure。为了否则返回一些默认值，你根本就

firstF(seq, predicate).recover { case _ => default } 

---

注1这里的代码演示了基本原则和计划内或计划外的异常之间没有区别。

注2此代码不是尾递归。