9

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

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

4

2 回答 2

7

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

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

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

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

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

您有机会将异常类型映射到 failure Directive，这从监督的角度定义了如何处理失败。选项包括：

  1. Stop - 完全停止子actor并且不再向它发送任何消息

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

  3. Restart - 类似于 resume，但在这种情况下，旧实例被丢弃并构造一个新实例并重置内部状态（preStart）

  4. 升级 - 将链向上升级到主管的父级

因此，假设给定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
于 2013-07-22T13:04:35.497 回答
0

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

  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 }
    }

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

于 2015-01-16T13:06:56.453 回答