Akka Future recover 可以恢复什么？

Question

我很好奇Future中的recover函数在什么情况下可以恢复抛出的异常？我正在同时使用 Akka Actor 和 Future：

这是我打未来电话的地方：

implicit val timeout = Timeout(5.seconds) //yes, I already have this line.

val response = (ActorA ? someMessage(someStuff))
                    .mapTo[TransOk]
                    .map(message => (OK, message.get))
                    .recover{
                    case e => (BadRequest, e.getMessage)
                  }

我发送ActorA，然后将结果映射到TransOK类，最后我添加.recover{}。

那么这是ActorA的方法：

case someMessage(stuff) =>
      //the exception being thrown here is not captured by Future.recover() method
      //why!?
      val id = if (some.canFind(stuff)) doSomething() 
               else throw new Exception("ERROR ERROR!")

      val result: Try[SomeDBType] = DAL.db.withSession { implicit session =>
        Try(DB.findStuff(stuff))
      }

      result match {
        case Success(content) => sender ! TransOk(content)
        case Failure(ex) => throw ex //let it escalate
      }

有趣的是：.recover() 没有捕获到第一个异常。那么recover在什么情况下能够捕获到异常呢？我认为它涵盖了正在调用的方法中发生的所有异常。

Answer 1

Future 中的 recover 函数设置为处理计算 Future 本身的值失败的情况。考虑以下情况：

val f:Future[Int] = future{
  val s:String = null
  s.length
}

在这种情况下，因为String 始终为空，所以此Future 将始终失败。如果我们想在不管失败的情况下始终为这个Future 赋值，我们会像这样使用recover：

val finalFut = f.recover{case ex => 1}

在这种情况下，如果我未来的计算失败，我总是有一个 Future 包装值 1。

现在有了 ask 和 Akka，我知道有两种方法可以让 ask Future 失败。第一个涉及发生超时。在这种情况下，您的recover 肯定会发挥作用。第二种情况涉及接收参与者将Status.Failure 向上传播给发送者，如下所示：

def receive = {
  case _ => sender ! Status.Failure(new RuntimeException("foo"))
}

这样做会导致基于上游请求的未来以您希望的方式失败，并导致您的恢复启动。如果参与者本身抛出未捕获的异常，那么主管将重新启动它（通常），但该异常默认情况下，除非您通过 Status.Failure 明确地做到这一点，否则不会从您的询问中向上游传播到未来。

Answer 2

ActorA 正在抛出异常，以便将错误升级到其主管。您永远不会发送回复以供 Future 处理。您可以尝试设置更短的超时时间，看看会得到什么。

此示例 scala 脚本将立即显示错误，但在 5 秒后显示错误消息，说明您遇到了超时而不是另一个错误：

import akka.actor.{Actor,ActorSystem,Props}
import akka.pattern.ask
import akka.util.Timeout
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

class ActorA() extends Actor {
  def receive = {
    case "message" => throw new Exception("oops")
  }
}

implicit val timeout = Timeout(5.seconds)

val sys = ActorSystem("sys")
val a = sys.actorOf(Props(new ActorA()))

(a ? "message").recover { case e => "bad things happened: " + e }  foreach (println)