如何使用询问模式和监督处理异常答案

【问题标题】：How to handle exception with ask pattern and supervision如何使用询问模式和监督处理异常
【发布时间】：2013-07-22 08:22:06
【问题描述】：

我应该如何处理这里 DbActor 抛出的异常？我不知道如何处理它，应该通过管道失败案例吗？

class RestActor extends Actor with ActorLogging {
  import context.dispatcher

  val dbActor = context.actorOf(Props[DbActor])
  implicit val timeout = Timeout(10 seconds)


  override val supervisorStrategy: SupervisorStrategy = {
    OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 10 seconds) {
      case x: Exception => ???
    }
  }

  def receive = {
    case GetRequest(reqCtx, id) => {

        // perform db ask
       ask(dbActor, ReadCommand(reqCtx, id)).mapTo[SomeObject] onComplete {
        case Success(obj) => { // some stuff }
        case Failure(err) => err match {
          case x: Exception => ???
        }
      }
    }
  }
}

很高兴得到您的想法，在此先感谢！

【问题讨论】：

标签： scala akka akka-supervision

【解决方案1】：

根据您的代码示例中的问题，我可以在这里看到几个问题：

当我在如何处理异常的定义中覆盖默认的主管行为时，我可以做哪些类型的事情？

使用ask 时，当我在等待的Future 上得到Failure 结果时，我可以做什么类型的事情？

让我们先从第一个问题开始（通常是个好主意）。当您覆盖默认的主管策略时，您可以更改子actor 中某些类型的未处理异常的处理方式，以及如何处理该失败的子actor。上一句中的关键词是unhandled。对于正在执行请求/响应的参与者，您实际上可能希望处理（捕获）特定异常并返回某些响应类型（或者使上游未来失败，稍后会详细介绍），而不是让它们不被处理。当发生未处理的异常时，您基本上无法用问题描述回复发件人，然后发件人可能会收到TimeoutException，因为他们的Future 永远不会完成。一旦弄清楚了要明确处理的内容，就可以在定义自定义主管策略时考虑所有其余的异常。在此块内：

OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 10 seconds) {
  case x: Exception => ???
}

您有机会将异常类型映射到故障Directive，它定义了从监督角度处理故障的方式。选项有：

停止 - 完全停止子actor并且不再向它发送任何消息
Resume - 恢复失败的子进程，而不是重新启动它，从而保持其当前的内部状态
Restart - 与 resume 类似，但在这种情况下，旧实例被丢弃，并构造一个新实例并重置内部状态（preStart）
升级 - 将链向上升级到主管的父级

假设给定 SQLException 您想恢复并给定所有其他人您想重新启动，那么您的代码将如下所示：

OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 10 seconds) {
  case x: SQLException => Resume
  case other => Restart
}

现在第二个问题与当Future 本身返回Failure 响应时要做什么有关。在这种情况下，我想这取决于Future 应该发生的事情。如果其余参与者自己负责完成 http 请求（假设 httpCtx 上有一个 complete(statusCode:Int, message:String) 函数），那么您可以执行以下操作：

   ask(dbActor, ReadCommand(reqCtx, id)).mapTo[SomeObject] onComplete {
    case Success(obj) => reqCtx.complete(200, "All good!")
    case Failure(err:TimeoutException) => reqCtx.complete(500, "Request timed out")
    case Failure(ex) => reqCtx.complete(500, ex.getMessage)
  }

现在，如果上游的另一个参与者负责完成 http 请求并且您需要响应该参与者，您可以执行以下操作：

   val origin = sender
   ask(dbActor, ReadCommand(reqCtx, id)).mapTo[SomeObject] onComplete {
    case Success(obj) => origin ! someResponseObject
    case Failure(ex) => origin ! Status.Failure(ex)
  }

这种方法假定在成功块中，您首先要在响应之前对结果对象进行按摩。如果您不想这样做并且想将结果处理推迟给发送者，那么您可以这样做：

   val origin = sender
   val fut = ask(dbActor, ReadCommand(reqCtx, id))
   fut pipeTo origin

【讨论】：

我想我明白了，这里真正的关键是了解未处理的内容

【解决方案2】：

对于更简单的系统，可能需要捕获并转发所有错误。为此，我制作了这个小函数来包装接收方法，而无需进行监督：

  import akka.actor.Actor.Receive
  import akka.actor.ActorContext
  /**
   * Meant for wrapping the receive method with try/catch.
   * A failed try will result in a reply to sender with the exception.
   * @example
   *          def receive:Receive = honestly {
   *            case msg => sender ! riskyCalculation(msg)
   *          }
   *          ...
   *          (honestActor ? "some message") onComplete {
   *            case e:Throwable => ...process error
   *            case r:_ => ...process result
   *          }
   * @param receive
   * @return Actor.Receive
   *
   * @author Bijou Trouvaille
   */
    def honestly(receive: =>Receive)(implicit context: ActorContext):Receive = { case msg =>
      try receive(msg) catch { case error:Throwable => context.sender ! error }
    }

然后您可以将其放入包文件中并导入 la akka.pattern.pipe 等。显然，这不会处理异步代码抛出的异常。

【讨论】：