Comment attendre plusieurs Futures ?

Question

Supposons que j'ai plusieurs contrats à terme et que je doive attendre jusqu'à ce que soit l'un d'entre eux échoue o tous réussissent.

Par exemple : Disons qu'il y a 3 contrats à terme : f1 , f2 , f3 .

• Si f1 réussit et f2 échoue je n'attends pas f3 (et retour échec au client).

• Si f2 échoue alors que f1 y f3 sont toujours en cours d'exécution, je ne les attends pas (et je retourne à la maison). échec )

• Si f1 réussit et ensuite f2 réussit, je continue à attendre f3 .

Comment la mettriez-vous en œuvre ?

Answer 1

Cette question a été répondue mais je poste ma solution de classe de valeur (les classes de valeur ont été ajoutées en 2.10) puisqu'il n'y en a pas ici. N'hésitez pas à critiquer.

  implicit class Sugar_PimpMyFuture[T](val self: Future[T]) extends AnyVal {
    def concurrently = ConcurrentFuture(self)
  }
  case class ConcurrentFuture[A](future: Future[A]) extends AnyVal {
    def map[B](f: Future[A] => Future[B]) : ConcurrentFuture[B] = ConcurrentFuture(f(future))
    def flatMap[B](f: Future[A] => ConcurrentFuture[B]) : ConcurrentFuture[B] = concurrentFutureFlatMap(this, f) // work around no nested class in value class
  }
  def concurrentFutureFlatMap[A,B](outer: ConcurrentFuture[A], f: Future[A] => ConcurrentFuture[B]) : ConcurrentFuture[B] = {
    val p = Promise[B]()
    val inner = f(outer.future)
    inner.future onFailure { case t => p.tryFailure(t) }
    outer.future onFailure { case t => p.tryFailure(t) }
    inner.future onSuccess { case b => p.trySuccess(b) }
    ConcurrentFuture(p.future)
  }

ConcurrentFuture est un wrapper Future sans surcharge qui change la carte Future/flatMap par défaut de do-this-then-that à combine-all-and-fail-if-any-fail. Utilisation :

def func1 : Future[Int] = Future { println("f1!");throw new RuntimeException; 1 }
def func2 : Future[String] = Future { Thread.sleep(2000);println("f2!");"f2" }
def func3 : Future[Double] = Future { Thread.sleep(2000);println("f3!");42.0 }

val f : Future[(Int,String,Double)] = {
  for {
    f1 <- func1.concurrently
    f2 <- func2.concurrently
    f3 <- func3.concurrently
  } yield for {
   v1 <- f1
   v2 <- f2
   v3 <- f3
  } yield (v1,v2,v3)
}.future
f.onFailure { case t => println("future failed $t") }

Dans l'exemple ci-dessus, f1, f2 et f3 s'exécuteront simultanément et si l'une d'entre elles échoue dans un ordre quelconque, le futur du tuple échouera immédiatement.

Answer 2

Vous pouvez également consulter l'API Future de Twitter. Notamment la méthode Future.collect. Elle fait exactement ce que vous voulez : https://twitter.github.io/scala_school/finagle.html

Le code source Future.scala est disponible ici : https://github.com/twitter/util/blob/master/util-core/src/main/scala/com/twitter/util/Future.scala

Answer 3

Vous pouvez utiliser ceci :

val l = List(1, 6, 8)

val f = l.map{
  i => future {
    println("future " +i)
    Thread.sleep(i* 1000)
    if (i == 12)
      throw new Exception("6 is not legal.")
    i
  }
}

val f1 = Future.sequence(f)

f1 onSuccess{
  case l => {
    logInfo("onSuccess")
    l.foreach(i => {

      logInfo("h : " + i)

    })
  }
}

f1 onFailure{
  case l => {
    logInfo("onFailure")
  }