Akka Decider 是否可以访问完整的失败场景？

Question

Akka 新手。创建一个扩展 SupervisorStrategy 的新 Scala 类为我提供了以下模板：

class MySupervisorStrategy extends SupervisorStrategy {
  override def decider: Decider = ???

  override def handleChildTerminated(context: ActorContext, child: ActorRef,
    children: Iterable[ActorRef]): Unit = ???

  override def processFailure(context: ActorContext, restart: Boolean,
    child: ActorRef, cause: Throwable, stats: ChildRestartStats, children: Iterable[ChildRestartStats]): Unit = ???
}

我正在寻找一种访问方式：

1. Throwable/Exception 是从儿童演员身上抛出的
2. 引发异常的子actor ActorRef
3. 传递给子actor的消息提示抛出异常

我认为只要孩子抛出异常，Decider（实际上是PartialFunction[Throwable,Directive]）就会通过Throwable，但我没有看到我可以访问的地方我上面的列表中的#2 和#3。 有什么想法吗？

---

更新

从发布的小提琴看来，一个有效的Decider 是：

{
    case ActorException(ref,t,"stop")      =>
      println(s"Received 'stop' from ${ref}")
      Stop
    case ActorException(ref,t,"restart")      =>
      println(s"Received 'restart' from ${ref}")
      Restart
    case ActorException(ref,t,"resume")      =>
      println(s"Received 'resume' from ${ref}")
      Resume
}

以上，我看到了所有三个：

1. 孩子抛出的异常
2. 引发异常的孩子 (ref)
3. 最初发送给孩子的消息（导致抛出异常）

看起来Decider 中没有需要在Supervisor 类中定义。我想将Decider 逻辑拉到MyDecider.scala 中，并找到一种方法来重构Supervisor，以便它的supervisorStrategy 使用MyDecider 的实例，所以可能类似于：

class Supervisor extends Actor {
  import akka.actor.OneForOneStrategy
  import akka.actor.SupervisorStrategy._
  import scala.concurrent.duration._

  var child: ActorRef = _

  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute, decider = myDecider)

  ...
}

Answer 1

对于#2，您可以访问发送者“如果策略在监督参与者内部声明”

如果策略是在监督参与者内部（而不是在伴随对象内）声明的，它的决策者可以以线程安全的方式访问参与者的所有内部状态，包括获取对当前失败的引用孩子（可作为失败消息的发送者）。

消息不可用，因此唯一的选择是捕获您的异常并使用收到的消息抛出自定义异常。

这是quick fiddle

class ActorSO extends Actor {

  def _receive: Receive = {
    case e =>
      println(e)
      throw new RuntimeException(e.toString)
  }

  final def receive = {
    case any => try {
      _receive(any)
    }
    catch {
      case t:Throwable => throw new ActorException(self,t,any)
    }

  }

}

更新

Decider 只是 PartialFunction，因此您可以在构造函数中传递它。

object SupervisorActor {
  def props(decider: Decider) = Props(new SupervisorActor(decider))
}

class SupervisorActor(decider: Decider) extends Actor {

  override val supervisorStrategy = OneForOneStrategy()(decider)

  override def receive: Receive = ???
}

class MyDecider extends Decider {
  override def isDefinedAt(x: Throwable): Boolean = true

  override def apply(v1: Throwable): SupervisorStrategy.Directive = {
    case t:ActorException => Restart
    case notmatched => SupervisorStrategy.defaultDecider.apply(notmatched)
  }
}

object Test {
  val myDecider: Decider = {
    case t:ActorException => Restart
    case notmatched => SupervisorStrategy.defaultDecider.apply(notmatched)
  }
  val myDecider2 = new MyDecider()
  val system = ActorSystem("stackoverflow")
  val supervisor = system.actorOf(SupervisorActor.props(myDecider))
  val supervisor2 = system.actorOf(SupervisorActor.props(myDecider2))
}

这样做，您将无法访问像通过sender() 引发异常的孩子的ActorRef 这样的主管状态（尽管我们将其包含在ActorException 中）

关于从主管访问导致异常的子消息的原始问题，您可以看到 here（来自 akka 2.5.3），akka 开发人员选择不使其可用于决策。

  final protected def handleFailure(f: Failed): Unit = {
    // ¡¡¡ currentMessage.message is the one that cause the exception !!!
    currentMessage = Envelope(f, f.child, system)
    getChildByRef(f.child) match {
      /*
       * only act upon the failure, if it comes from a currently known child;
       * the UID protects against reception of a Failed from a child which was
       * killed in preRestart and re-created in postRestart
       */
      case Some(stats) if stats.uid == f.uid ⇒
        // ¡¡¡ currentMessage.message is not passed to the handleFailure !!!
        if (!actor.supervisorStrategy.handleFailure(this, f.child, f.cause, stats, getAllChildStats)) throw f.cause
      case Some(stats) ⇒
        publish(Debug(self.path.toString, clazz(actor),
          "dropping Failed(" + f.cause + ") from old child " + f.child + " (uid=" + stats.uid + " != " + f.uid + ")"))
      case None ⇒
        publish(Debug(self.path.toString, clazz(actor), "dropping Failed(" + f.cause + ") from unknown child " + f.child))
    }
  }