博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
[翻译]AKKA笔记 -ACTOR SUPERVISION - 8
阅读量:5803 次
发布时间:2019-06-18

本文共 14798 字,大约阅读时间需要 49 分钟。

失败更像是分布式系统的一个特性。因此Akka用一个容忍失败的模型,在你的业务逻辑与失败处理逻辑(supervision逻辑)中间你能有一个清晰的边界。只需要一点点工作,这很赞。这就是我们要讨论的主题。

ACTOR SUPERVISION

想象一个方法调用了你栈顶的方法但却出了一个异常。那么在栈下的方法能做什么呢?

  1. 抓住异常并按顺序处理恢复

  2. 抓住异常,也许记个日志并保持安静。

  3. 下层的方法也可以选择无视这个异常(或者抓住并重扔出来)

想象下一直扔到main方法仍然没有处理这个异常。这种情况下,程序肯定会输出一个异常给console后退出。

图片描述

你可以把同样的情况套用在线程上。如果一个子线程抛了异常而再假设run*call方法没有处理它,那么这个异常就会期望放在父线程或主线程中解决,无论哪种情况,如果主线程没有处理他,系统就会退出。

让我们再看看 - 如果被context.actorof创建出来的子Actor因为一个异常失败了。父actor(指supervisor)可以处理子actor的任何失败。如果父actor做了,他可以处理并恢复(Restart/Resume)。另外,把异常传递(Escalate)给父actor。 还有一种做法,可以直接stop掉子actor - 这就是那个子actor结局了。 为什么我说父actor(那个supervisor)?这是因为akka的监护方式为家长监护 - 这意味着只有创建了actor的人才能监护他们。

就这么多了!我们已经覆盖到所有监护指令(Directoives)了。

策略

我忘了说一点: 你已经知道一个Akka Actor可以创建子actor并且子actor也可以随意创建他们自己的子actor。

现在,想下以下两个场景:

1.OneForOneStrategy

你的actor创建了很多子actor并且每一个子actor都连接了不同的数据源。假设你运行的是一个将英语翻译成多种语言的应用。

图片描述

假设,一个子actor失败了然而你可以接受在最终结果里跳过这个结果,你想怎么做?关掉这个服务?当然不,你可能想要重启/关闭这个有问题的子actor。是吧?现在这个策略在Akka的监护策略中叫OneForOneStrategy策略 - 如果一个actor挂了,只单独处理这个actor。

基于你的业务异常,你可能需要对不同的异常有不同的反应(停止,重启,升级,恢复)。要配置你自己的策略,你只需要override你Actor类中的supervisorStrategy方法。

声明OneForOneStrategy的例子

import akka.actor.Actor  import akka.actor.ActorLogging  import akka.actor.OneForOneStrategy  import akka.actor.SupervisorStrategy.Stopclass TeacherActorOneForOne extends Actor with ActorLogging {    ...    ...  override val supervisorStrategy=OneForOneStrategy() {    case _: MinorRecoverableException     => Restart    case _: Exception                   => Stop  }  ...  ...

2.AllForOneStrategy策略

假设你在做一个 (这是个又证明了我没啥创造力的例子!),你的每个块都被一个不同的actor处理。突然,一个Actor失败了并抛了一个异常。这样再往下处理就没什么意思了因为最终结果肯定是错的。所以,逻辑就是停止stop所有的actor。

图片描述

我为什么说stop而不是重启?因为在这个例子里重启也没用,每个actor的mailbox在重启时并不会被清理。所以,如果我们重启了,另外的chunk仍然会被处理。这不是我们想要的。重建actor并用新的mailbox在这里是个更合适的策略。

OneForOneStrategy一样,只需要用AllForOneStrategy的实现覆写supervisorStrategy

下面是例子

import akka.actor.{Actor, ActorLogging}  import akka.actor.AllForOneStrategy  import akka.actor.SupervisorStrategy.Escalate  import akka.actor.SupervisorStrategy.Stopclass TeacherActorAllForOne extends Actor with ActorLogging {  ...  override val supervisorStrategy = AllForOneStrategy() {    case _: MajorUnRecoverableException => Stop    case _: Exception => Escalate  }  ...  ...

指令 DIRECTIVES

AllForOneStrategyOneForOneStrategy的构造方法都接受一个叫DeciderPartialFunction[Throwable,Directive]方法,他把ThrowableDirective指令做了一个映射:

case _: MajorUnRecoverableException => Stop

这就简单的四个指令 - Stop,Resume,Escalate和Restart

Stop

在异常发生时子actor会停止,任何发给停止的actor的消息都会被转到deadLetter队列。

Resume

子actor会忽略抛出异常的消息并且继续处理队列中的其他消息。

Restart

子actor会停止并且一个新的actor会初始化。继续处理mailbox中其他的消息。世界对这个是无感知的因为同样的ActorRef指向了新的Actor。

Escalate

supervisor复制了失败并让他的supervisor处理这个异常。

缺省策略

如果我们的actor没指定任何策略但是创建了子actor。他们会怎样处理?Actor会有一个缺省的策略:

override val supervisorStrategy=OneForOneStrategy() {    case _: ActorInitializationException=> Stop    case _: ActorKilledException        => Stop    case _: DeathPactException             => Stop    case _: Exception                   => Restart}

所以,缺省策略处理了四个case:

1. ACTORINITIALIZATIONEXCEPTION => STOP

当actor不能初始化,他会抛出一个ActorInitializationException。actor会被停止。让我们在preStart调用中模拟下这个:

package me.rerun.akkanotes.supervisionimport akka.actor.{ActorSystem, Props}  import me.rerun.akkanotes.protocols.TeacherProtocol.QuoteRequest  import akka.actor.Actor  import akka.actor.ActorLoggingobject ActorInitializationExceptionApp extends App{  val actorSystem=ActorSystem("ActorInitializationException")  val actor=actorSystem.actorOf(Props[ActorInitializationExceptionActor], "initializationExceptionActor")  actor!"someMessageThatWillGoToDeadLetter"}class ActorInitializationExceptionActor extends Actor with ActorLogging{    override def preStart={    throw new Exception("Some random exception")  }  def receive={    case _=>  }}Ru

运行ActorInitializationExceptionApp会产生一个ActorInitializationException 异常然后所有的消息都会进deadLetterActor的消息队列:

Log

[ERROR] [11/10/2014 16:08:46.569] [ActorInitializationException-akka.actor.default-dispatcher-2] [akka://ActorInitializationException/user/initializationExceptionActor] Some random exceptionakka.actor.ActorInitializationException: exception during creation      at akka.actor.ActorInitializationException$.apply(Actor.scala:164)......Caused by: java.lang.Exception: Some random exception      at me.rerun.akkanotes.supervision.ActorInitializationExceptionActor.preStart(ActorInitializationExceptionApp.scala:17)......[INFO] [11/10/2014 16:08:46.581] [ActorInitializationException-akka.actor.default-dispatcher-4] [akka://ActorInitializationException/user/initializationExceptionActor] Message [java.lang.String] from Actor[akka://ActorInitializationException/deadLetters] to Actor[akka://ActorInitializationException/user/initializationExceptionActor#-1290470495] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

2. ACTORKILLEDEXCEPTION => STOP

当Actor被kill消息关闭后,他会抛出一个ActorKilledException。如果抛这个异常,缺省策略会让子actor停止。看起来停止一个被kill掉的actor没什么意义。但想想这个:

  1. ActorKilledException 会被传递给supervisor。 那么之前我们在里面提到的Actor里的生命周期watchdeathwatchers。 直到Actor被停掉前watcher不会知道任何事情。

  2. 给Actor发送kill只会让那个特定的监管actor知道。用stop处理会暂停那个actor的mailbox,暂停了子actor的mailbox,停止了子actor,发送了Terminated给所有子actor的watcher,发送给所有类一个Terminated,然后actor的watcher都会迅速失败最终让Actor自己停止、

package me.rerun.akkanotes.supervisionimport akka.actor.{ActorSystem, Props}  import me.rerun.akkanotes.protocols.TeacherProtocol.QuoteRequest  import akka.actor.Actor  import akka.actor.ActorLogging  import akka.actor.Killobject ActorKilledExceptionApp extends App{  val actorSystem=ActorSystem("ActorKilledExceptionSystem")  val actor=actorSystem.actorOf(Props[ActorKilledExceptionActor])  actor!"something"  actor!Kill  actor!"something else that falls into dead letter queue"}class ActorKilledExceptionActor extends Actor with ActorLogging{    def receive={    case message:String=> log.info (message)  }}

Log

日志说只要ActorKilledException 进来,supervisor就会停掉actor并且消息会进入deadLetter队列

INFO  m.r.a.s.ActorKilledExceptionActor - somethingERROR akka.actor.OneForOneStrategy - Kill  akka.actor.ActorKilledException: KillINFO  akka.actor.RepointableActorRef - Message [java.lang.String] from Actor[akka://ActorKilledExceptionSystem/deadLetters] to Actor[akka://ActorKilledExceptionSystem/user/$a#-1569063462] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

3. DEATHPACTEXCEPTION => STOP

在[DeathWatch]()文中,你可以看到当一个Actor观察一个子Actor时,他期望在他的receive中处理Terminated消息。如果没有呢?你会得到一个DeathPactException

图片描述

代码演示了supervisorwatch子actor但没有从子actor处理Terminated消息。

package me.rerun.akkanotes.supervisionimport akka.actor.{ActorSystem, Props}  import me.rerun.akkanotes.protocols.TeacherProtocol.QuoteRequest  import akka.actor.Actor  import akka.actor.ActorLogging  import akka.actor.Kill  import akka.actor.PoisonPill  import akka.actor.Terminatedobject DeathPactExceptionApp extends App{  val actorSystem=ActorSystem("DeathPactExceptionSystem")  val actor=actorSystem.actorOf(Props[DeathPactExceptionParentActor])  actor!"create_child" //Throws DeathPactException  Thread.sleep(2000) //Wait until Stopped  actor!"someMessage" //Message goes to DeadLetters}class DeathPactExceptionParentActor extends Actor with ActorLogging{  def receive={    case "create_child"=> {      log.info ("creating child")      val child=context.actorOf(Props[DeathPactExceptionChildActor])      context.watch(child) //Watches but doesnt handle terminated message. Throwing DeathPactException here.      child!"stop"    }    case "someMessage" => log.info ("some message")    //Doesnt handle terminated message    //case Terminated(_) =>  }}class DeathPactExceptionChildActor extends Actor with ActorLogging{    def receive={    case "stop"=> {      log.info ("Actor going to stop and announce that it's terminated")      self!PoisonPill    }  }}

Log

日志告诉我们DeathPactException 进来了,supervisor停止了actor然后消息都进入了deadLetter的队列

INFO  m.r.a.s.DeathPactExceptionParentActor - creating childINFO  m.r.a.s.DeathPactExceptionChildActor - Actor going to stop and announce that it's terminatedERROR akka.actor.OneForOneStrategy - Monitored actor [Actor[akka://DeathPactExceptionSystem/user/$a/$a#-695506341]] terminated  akka.actor.DeathPactException: Monitored actor [Actor[akka://DeathPactExceptionSystem/user/$a/$a#-695506341]] terminatedINFO  akka.actor.RepointableActorRef - Message [java.lang.String] from Actor[akka://DeathPactExceptionSystem/deadLetters] to Actor[akka://DeathPactExceptionSystem/user/$a#-1452955980] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

4. EXCEPTION => RESTART

对于其他的异常,缺省的指令重启Actor。看下这个应用。只是要证明下Actor重启了,OtherExceptionParentActor让child抛出一个异常并立刻发送一条消息。消息在子actor重启的时候到达了mailbox,并被处理了。真不错!

图片描述

package me.rerun.akkanotes.supervisionimport akka.actor.Actor  import akka.actor.ActorLogging  import akka.actor.ActorSystem  import akka.actor.OneForOneStrategy  import akka.actor.Props  import akka.actor.SupervisorStrategy.Stopobject OtherExceptionApp extends App{  val actorSystem=ActorSystem("OtherExceptionSystem")  val actor=actorSystem.actorOf(Props[OtherExceptionParentActor])  actor!"create_child"}class OtherExceptionParentActor extends Actor with ActorLogging{  def receive={    case "create_child"=> {      log.info ("creating child")      val child=context.actorOf(Props[OtherExceptionChildActor])      child!"throwSomeException"      child!"someMessage"    }  }}class OtherExceptionChildActor extends akka.actor.Actor with ActorLogging{  override def preStart={    log.info ("Starting Child Actor")  }  def receive={    case "throwSomeException"=> {      throw new Exception ("I'm getting thrown for no reason")     }    case "someMessage" => log.info ("Restarted and printing some Message")  }  override def postStop={    log.info ("Stopping Child Actor")  }}

Log

1.异常抛出了,我们能在trace中看到

  1. 子类重启了 - stop和start被调用了(我们稍后能看到preRestart和postRestart

  2. 消息在重启开始前被发送给子actor。

INFO  m.r.a.s.OtherExceptionParentActor - creating childINFO  m.r.a.s.OtherExceptionChildActor - Starting Child ActorERROR akka.actor.OneForOneStrategy - I'm getting thrown for no reasonjava.lang.Exception: I'm getting thrown for no reason      at me.rerun.akkanotes.supervision.OtherExceptionChildActor$$anonfun$receive$2.applyOrElse(OtherExceptionApp.scala:39) ~[classes/:na]    at akka.actor.Actor$class.aroundReceive(Actor.scala:465) ~[akka-actor_2.11-2.3.4.jar:na]......    INFO  m.r.a.s.OtherExceptionChildActor - Stopping Child ActorINFO  m.r.a.s.OtherExceptionChildActor - Starting Child ActorINFO  m.r.a.s.OtherExceptionChildActor - Restarted and printing some Message

ESCALATE AND RESUME

我们在defaultStrategy中看到stop和restart的例子。现在让我们快速看下Escalate

Resume忽略了异常并处理mailbox中的下条消息。这就像是抓住了异常但什么事也没做。

Escalating更像是异常是致命的而supervisor不能处理它。所以,他要向他的supervisor求救。让我们看个例子。

假设有三个Actor - EscalateExceptionTopLevelActor, EscalateExceptionParentActor 和 EscalateExceptionChildActor。 如果一个子actor抛出一个日常并且父级别actor不能处理它,他可以Escalate这个异常到顶级actor。顶级actor也可以选择对哪些指令做出响应。在我们的例子里,我们只是做了stopstop会立即停掉child(这里是EscalateExceptionParentActor)。我们知道,当一个actor执行stop时,他的所有子类都会在actor自己停掉前先停止。

图片描述

package me.rerun.akkanotes.supervisionimport akka.actor.Actor  import akka.actor.ActorLogging  import akka.actor.ActorSystem  import akka.actor.OneForOneStrategy  import akka.actor.Props  import akka.actor.SupervisorStrategy.Escalate  import akka.actor.SupervisorStrategy.Stop  import akka.actor.actorRef2Scalaobject EscalateExceptionApp extends App {  val actorSystem = ActorSystem("EscalateExceptionSystem")  val actor = actorSystem.actorOf(Props[EscalateExceptionTopLevelActor], "topLevelActor")  actor ! "create_parent"}class EscalateExceptionTopLevelActor extends Actor with ActorLogging {  override val supervisorStrategy = OneForOneStrategy() {    case _: Exception => {      log.info("The exception from the Child is now handled by the Top level Actor. Stopping Parent Actor and its children.")      Stop //Stop will stop the Actor that threw this Exception and all its children    }  }  def receive = {    case "create_parent" => {      log.info("creating parent")      val parent = context.actorOf(Props[EscalateExceptionParentActor], "parentActor")      parent ! "create_child" //Sending message to next level    }  }}class EscalateExceptionParentActor extends Actor with ActorLogging {  override def preStart={    log.info ("Parent Actor started")  }  override val supervisorStrategy = OneForOneStrategy() {    case _: Exception => {      log.info("The exception is ducked by the Parent Actor. Escalating to TopLevel Actor")      Escalate    }  }  def receive = {    case "create_child" => {      log.info("creating child")      val child = context.actorOf(Props[EscalateExceptionChildActor], "childActor")      child ! "throwSomeException"    }  }  override def postStop = {    log.info("Stopping parent Actor")  }}class EscalateExceptionChildActor extends akka.actor.Actor with ActorLogging {  override def preStart={    log.info ("Child Actor started")  }  def receive = {    case "throwSomeException" => {      throw new Exception("I'm getting thrown for no reason.")    }  }  override def postStop = {    log.info("Stopping child Actor")  }}

Log

可以在log中看到,

  1. 子actor抛了异常。

  2. supervisor(EscalateExceptionParentActor)升级了(escalate)异常抛给了他的supervisor(EscalateExceptionTopLevelActor

  3. EscalateExceptionTopLevelActor 的指令是关闭actor。在顺序上,子actor先停止。

  4. 父actor之后再停止(在watcher被通知后)

INFO  m.r.a.s.EscalateExceptionTopLevelActor - creating parentINFO  m.r.a.s.EscalateExceptionParentActor - Parent Actor startedINFO  m.r.a.s.EscalateExceptionParentActor - creating childINFO  m.r.a.s.EscalateExceptionChildActor - Child Actor startedINFO  m.r.a.s.EscalateExceptionParentActor - The exception is ducked by the Parent Actor. Escalating to TopLevel ActorINFO  m.r.a.s.EscalateExceptionTopLevelActor - The exception from the Child is now handled by the Top level Actor. Stopping Parent Actor and its children.ERROR akka.actor.OneForOneStrategy - I'm getting thrown for no reason.  java.lang.Exception: I'm getting thrown for no reason.      at me.rerun.akkanotes.supervision.EscalateExceptionChildActor$$anonfun$receive$3.applyOrElse(EscalateExceptionApp.scala:71) ~[classes/:na]    ...    ...INFO  m.r.a.s.EscalateExceptionChildActor - Stopping child ActorINFO  m.r.a.s.EscalateExceptionParentActor - Stopping parent Actor

请记住无论哪个指令发出只会使用在被escalated的子类上。 例如,一个restart指令从顶层发出,只有父类会被重启并且在构造函数中/preStart中的都会被执行。如果一个父actor的子类在构造函数中呗创建,他们就会被创建。然而,在消息中创建的child的父actor仍然会在Terminated状态。

TRIVIA

实际上,你可以控制是否preStart被调用。我们可以在下节看到。如果你好奇,可以看下中的*postRestart方法

def postRestart(reason: Throwable): Unit = {    preStart()}

代码

跟往常一样,代码在


文章来自微信平台「麦芽面包」(微信扫描二维码关注)。未经允许,禁止转载。

图片描述

你可能感兴趣的文章
线性表基本维护[ACM]
查看>>
安卓平台基于SIP协议实现注册,聊天功能
查看>>
Java_spark简单例子
查看>>
imshow(K)和imshow(K,[]) 的区别
查看>>
poj3190 Stall Reservations
查看>>
CORS 跨域问题, 以及作为api server 的正确配置, 后台 nginx 配置
查看>>
loadrunner录制脚本、回放脚本遇到的问题
查看>>
16进制数至字符串转换
查看>>
Java Web整合开发(13) -- XML
查看>>
标准库Queue的实现
查看>>
adb shell dumpsys的使用
查看>>
CentOS 7 安装后没有ifconfig命令
查看>>
装饰器进阶
查看>>
webbrowser 请求的资源在使用中。 (异常来自 HRESULT:0x800700AA)
查看>>
如何使用Python3.4连接MySQL
查看>>
automake,autoconf使用详解
查看>>
高并发
查看>>
(转载)Attempting to add QLayout "" to MainWindow "", which already has a layout
查看>>
CentOS 6.5 Rsync+Inotify实时同步
查看>>
DevCloud会员权益升级!日常领码豆,轻松换好礼!
查看>>