Scala未来的失败是为了理解

Question

我有三个顺序期货,并在这样的理解中使用

val comF = for {
  f1 <- future1
  f2 <- future2
  f3 <- future3
} yield {
  // something
}

comF onSuccess { }
comF onFailure { 
  // ----------------      Here is the problem  --------------------------------
  //
  // How do I know which future failed(throw exception), when the callback comes here ?
  // Thanks for the help! Different futures using different exceptions can solve it.
}

现在我有一个像List [Future [T]]这样的未来列表,首先我使用这种方法将它转移到Future [List [T]](为什么这个未来的列表转换为未来的列表转换编译和工作？).然后我得到了未来

val fList: Future[List[T]]
fList on Failure {
  // 
  // How do I know which is Fail now >??
}

Answer 1

考虑一下代码:

def func = {
  try {
    val x = maybeThrows
    val y = maybeThrowsToo
    val z = maybeThrowsAsWell
    result(x, y, x)
  } catch (RuntimeException e) {
    // How do I know which maybeThrows failed?
  }
}

该Future案件的工作基本上以同样的方式.

甚至分组计算List也无济于事:

def func = {
  try {
    val x = maybeThrows
    val y = maybeThrowsToo
    val z = maybeThrowsAsWell
    val list = List(x, y, z)
    result(list)
  } catch (RuntimeException e) {
    // How do I know which maybeThrows failed?
  }
}

Spoiler:你必须跟踪explicityl哪个计算失败.如果完成,它将导致一些样板try/catch.但幸运的是,Future(和Try)样板文件并没有那么糟糕:

class TaggedException(val idx, exc: Exception)

def tagFailedWithIndex[T](idx: Int, f: Future[T]): Future[T] = 
  future recoverWith { case exc => Future.failed(new TaggedException(idx, exc)) }

val comF = for {
  f1 <- tagFailedWithIndex(0, future1)
  f2 <- tagFailedWithIndex(1, future2)
  f3 <- tagFailedWithIndex(2, future3)
} yield something(f1, f2, f3)

comF onFailure { 
  case exc: TaggedException => "%d computation failed".format(exc.idx)
}

Spoiler你必须明确跟踪哪个计算失败.如果完成,它将导致很多样板try/catch.但幸运的是Try,并且Future行为更加相同:

class TaggedException(val idx, exc: Exception)

def tagFailedWithIndex[T](idx: Int, f: Future[T]): Future[T] = 
  future recoverWith { case exc => Future.failed(new TaggedException(idx, exc)) }

val comF = for {
  f1 <- tagFailedWithIndex(0, future1)
  f2 <- tagFailedWithIndex(1, future2)
  f3 <- tagFailedWithIndex(2, future3)
} yield something(f1, f2, f3)

comF onFailure { 
  case exc: TaggedException => "%d computation failed".format(exc.idx)
}

Answer 2

问题：flatMap 将 Futures 合并为一个 Future

您正在使用flatMap，因此期货嵌套在一个期货中。

你的代码是

import scala.concurrent.Future

val future1, future2, future3 = Future[Any]()

val comF = for {
  f1 <- future1
  f2 <- future2
  f3 <- future3
} yield {
  // something
}

comF onSuccess { ??? }
comF onFailure { ??? }

当我将脱糖规则应用于你的理解时，我得到

val comF = (future1).flatMap { case f1 => 
  (future2).flatMap { case f2 => 
    (future3).map { case f3 => {
        // something
      }
    }
  }
}

你可以清楚地看到flatMap这里的用法。原则上flatMap做两件事：它将函数应用于第一个Future. 这必须是一个将第一个的结果映射Future到另一个的函数Future，即Future嵌套在第一个Future（这是map部分）中。然后它“取消嵌套”这两个Futures并将它们合并为一个Future（flat部分）。在这一点上，两者Futures不再存在（从概念的角度来看；从技术上讲，它们仍然存在）。相反，只有一个 'merged' Future。

这两个调用从原来的三个调用flatMap创建一个新Future的。这就是为什么您无法找出三个原始文件中Futures的哪一个引发异常的原因：它们都不是。相反，新创建的Future运行并引发异常。

解决方案：单独跟踪您的进度

如果您想知道在引发异常之前计算的哪些步骤运行，您必须单独跟踪进度，例如通过Exception. 或者，您可以一个接一个地删除flatMap并运行单个Futures。

关于如何跟踪进度的示例：

首先我们创建一个新的异常类，它包含真正的异常和一些信息，异常来自哪里

class ExceptionWithOrigin[T](val origin : T, nested : Throwable) extends Exception(nested)

object ExceptionWithOrigin {
  def wrapFuture[T,U](origin : T, f : Future[U]) : Future[U] = {
    f.transform(Predef.identity, new ExceptionWithOrigin[T](origin,_))
  }

  def wrapFuture[U](f : Future[U]) = wrapFuture(f,f)
}

对于Futures我们没有特殊要求。

val future1,future2,future3 = Future[Any]()

然后我们Futures使用来自我们新创建的异常类的伴随对象的 helper 方法来包装给定的。

import ExceptionWithOrigin._

val comF = for {
  result1 <- wrapFuture(future1)
  result2 <- wrapFuture(future2)
  result3 <- wrapFuture(future3)
} yield {
  // something
}

当您捕获一些异常时ex，您现在可以使用ex.origin它来找出它的来源。当然，起源并不完全正确。原来的Futures future1，future2并future3没有真正执行。相反，新创建的Future运行，由flatMap. 但尽管如此，起源仍然有效。

关于更好命名的提示

顺便说一句，您应该将f1、f2和重命名f3为result1、result2和result3。它们不代表 a Future，而是每个的计算结果Future（每个Future返回的值）。