scala - 如何通过提问方式和监督来处理异常

Question

我应该如何处理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 => ???
        }
      }
    }
  }
}

希望得到您的想法，在此先感谢!

Answer 1

根据您的代码示例中的问题，我可以在此处看到几个问题:

1. What types of things can I do when I override the default supervisor behavior in the definition of how to handle exceptions?

2. When using ask, what types of things can I do when I get a Failure result on the Future that I am waiting on?

让我们首先从第一个问题开始(通常是一个好主意)。当您覆盖默认的主管策略时，您将具有更改处理该失败的子actor时如何处理子actor中某些类型的未处理异常的能力。上一句话中的关键字是 unhandled。对于正在执行请求/响应的参与者，您实际上可能想处理(捕获)特定的异常并返回某些响应类型(或者使上游的 future 失败，更多情况会在稍后进行)，而不是让它们不处理。当发生未处理的异常时，您基本上会失去用问题描述来响应发件人的能力，发件人可能会得到一个 TimeoutException而不是，因为他们的 Future将永远不会完成。一旦确定了要明确处理的内容，即可在定义自定义主管策略时考虑所有其他异常(exception)情况。在此块内:

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

您有机会将异常类型映射到故障 Directive，该代码从监管的角度定义了如何处理故障。选项包括:

Stop-完全停止子actor，并且不再向其发送任何消息

Resume-恢复失败的 child ，不重新启动它，从而保持其当前内部状态

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