Scala期货-如何从两个期货中获得结果或失败

Question

我使用 for 并行运行 2 个期货。我想知道在所有情况下哪个成功了，哪个失败了（所有情况都应该运行到完成，并显示结果或失败状态）。目前，我只能检索到综合成功结果

我从这里开始操作，但这还不够，因为我无法获得单个失败时的成功状态，也无法获得两个失败时的失败状态。在Scala未来的理解中，两个失败都是失败的

case class TaggedException(context:String, val throwable: Throwable) extends Exception(throwable.getMessage)

val f1 = Future {...}.recoverWith {case e:Throwable => Future.Failed(new TaggedException("first one failed", e))}
val f2 = Future {...}.recoverWith {case e: Throwable => Future.Failed(new TaggedException("second one failed", e))}

val combinedResult = for {
  r1 <- f1
  r2 <- f2
} yield (r1,r2)

combinedResult.onFailure {
case e : TaggedException => ... // if both fail I only get the first line in the for
// in case where single fails I only know fail status without the success of the second
}

我试图避免这种混乱:

var countCompleted = 0 ... or some other atomic way to count 
f1 onComplete {
  case Success(value) => {
    ... countCompleted increment ... 
    // handle success
    if both completed {
       // handle returning a status
    }
  }
  case Failure(error) => {
    ... countCompleted increment ... 
    // handle failure
    if both completed {
       // handle returning a status
    }
  }
}

f2 onComplete {
  case Success(value) => {
    ... countCompleted increment ... 
    // handle success
    if both completed {
       // handle returning a status
    }
  }
  case Failure(error) => {
    ... countCompleted increment ... 
    // handle failure
    if both completed {
       // handle returning a status
    }
  }
}

编辑:另一个版本-这是一个有效的方法吗？

def toFutureTry[A](future: Future[A]):Future[Try[A]] = future.map(Success(_)).recover {case t: Throwable => Failure(t)}

    val fa: Future[Try[Blah]] = toFutureTry(f1)
    val fb: Future[Try[Foo]] = toFutureTry(f2)

    val combinedRes = for {
      ra <- fa
      rb <- fb
    } yield (ra,rb)

    combinedRes.onComplete {
      case Success(successRes: (Try[Blah], Try[Foo])) => // all of these cases are success or fails
      case Failure(f: Throwable) => // i think it is unused right?
    }

Answer 1

对于理解，一旦一行失败，代码就会停止在那里并抛出任何异常。如果r1

我个人会把它们都放在＜code＞中，或者[可丢弃的，随便什么]，而不是把它们都放到＜code＞里。在将来使用进行恢复。失败（…）。通过这种方式，您可以使用您获得的任何左值执行操作，而不仅仅是执行onFailure，还可以使用您得到的任何右值执行其他操作。或者您可以使用＜code＞Try/Success/Failure＜/code＞…但这取决于您想如何处理错误。

我不知道您的具体用例，但如果您想单独处理每个成功或失败的案例，您可以这样做:

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.util.{Failure, Success, Try}

val f1 = Future(Try(1 / 1))
val f2 = Future(Try(1 / 0))

// for-comprehensions won't fall over when reading a Failure
// as failures don't extend Throwable
val combinedResult = for {
  r1 <- f1 // Success(1)
  r2 <- f2 // Failure(java.lang.ArithmeticException: / by zero)
} yield List(r1,r2)

combinedResult.map { // get inside Future
  f =>
    f.map { // get inside List
      case Success(a) => // do something with success
      case Failure(e: IndexOutOfBoundsException) => // do something with failure
      case Failure(e: ArithmeticException) => // do something with failure
      case Failure(e) => // do something with failure
    }
}

我个人不喜欢使用onComplete我更喜欢通过映射将数据尽可能长地保存在未来中。不过这只是个人喜好。

Answer 2

在Future[A]上使用FLMap不会有帮助，因为它总是会在其中一个产生的第一个故障时短路，您真的想在其中积累错误。

使用 Future.traverse 的解决方案，它将适用于任意多个 Future[A] 实例：

val f1 = Future.failed[Int](new Exception("42")).recoverWith {
  case e: Throwable => Future.failed(TaggedException("first one failed", e))
}

val f2 = Future(42).recoverWith {
  case e: Throwable =>
    Future.failed(TaggedException("second one failed", e))
}

val res: Future[List[Either[Throwable, Int]]] = 
  Future
   .traverse(List(f1, f2)) {
      eventualInt => eventualInt
       .map(i => Right(i))
       .recover { case err => Left(err) }
   }

res.onComplete {
  case Failure(exception) =>
    println(exception)
  case Success(value) =>
    value.foreach {
      case Right(int) => println(s"Received num: $int")
      case Left(err) => println(s"Oh no, err: $err")
    }
}

Await.result(res, Duration.Inf)

我们还可以使用猫对其验证类型的一点帮助：

import cats.data.Validated.{Invalid, Valid}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}
import cats.implicits._
import scala.util.{Failure, Success}

def main(args: Array[String]): Unit = {
  case class TaggedException(context: String, throwable: Throwable)
    extends Exception(throwable.getMessage)

  val f1 = Future.failed[Int](new Exception("42")).recoverWith {
    case e: Throwable => Future.failed(TaggedException("first one failed", e))
  }

  val f2 = Future(42).recoverWith {
    case e: Throwable => Future.failed(TaggedException("second one failed", e))
  }

  val res: Future[List[Validated[Throwable, Int]]] = 
    List(f1, f2)
     .traverse(eventualInt => eventualInt
                       .map(i => Valid(i))
                       .recover { case err => Invalid(err) })

  res.onComplete {
    case Failure(exception) =>
      println(exception)
    case Success(value) =>
      value.foreach {
        case Valid(int) => println(s"Received num: $int")
        case Invalid(err) => println(s"Oh no, err: $err")
      }
  }

  Await.result(res, Duration.Inf)
}

将屈服：

Oh no, err: TaggedException$3: 42
Received num: 42

Answer 3

您可以像这样组合< code>transform和< code>zip:

val combinedResult: Future[(Try[T], Try[T])] =
  f1.transform(Success(_)).zip(f2.transform(Success(_)))

然后你可以做:

combinedResult map {
  case (Success(v1), Success(v2)) =>
  case (Success(v1), Failure(f2)) =>
  case (Failure(f1), Success(v2)) =>
  case (Failure(f1), Failure(f2)) =>
}