akka actor父子监管的实现

akka中，父actor可以定义supervisorStartegy来实现对子actor的异常监管应对策略。

*   override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
*     case _: ArithmeticException      => Resume
*     case _: NullPointerException     => Restart
*     case _: IllegalArgumentException => Stop
*     case _: Exception                => Escalate
*   }

以上为一个监管策略的具体实现，可以看到具体的异常应对策略有Resume，Restart，Stop和Escalate。

当父actor生成子actor的时候，将会通过在子actor中的parent记录父actor来传递异常信息。

下面是具体的监管实现。

在子actor处理消息的invoke()函数中，将会捕获在receive()函数中抛出的异常，并在handleInvokeFailure()函数中进行处理。

final def invoke(messageHandle: Envelope): Unit = {val influenceReceiveTimeout = !messageHandle.message.isInstanceOf[NotInfluenceReceiveTimeout]try {currentMessage = messageHandleif (influenceReceiveTimeout)cancelReceiveTimeout()messageHandle.message match {case _: AutoReceivedMessage => autoReceiveMessage(messageHandle)case msg                    => receiveMessage(msg)}currentMessage = null // reset current message after successful invocation} catch handleNonFatalOrInterruptedException { e =>handleInvokeFailure(Nil, e)} finally {// Schedule or reschedule receive timeoutcheckReceiveTimeout(reschedule = influenceReceiveTimeout)}
}

在handleInvokeFailure()函数中，将会直接将错误信息封装成Failed类型的系统消息投递到parent父actor的系统消息队列中。

父actor将会通过读取系统消息队列的时候读取到子actor中发生的异常。

在systemInvoke中的invokeAll()函数中将会进行具体处理。

@tailrec
def invokeAll(messages: EarliestFirstSystemMessageList, currentState: Int): Unit = {val rest = messages.tailval message = messages.headmessage.unlink()try {message match {case message: SystemMessage if shouldStash(message, currentState) => stash(message)case f: Failed                                                    => handleFailure(f)case DeathWatchNotification(a, ec, at)                            => watchedActorTerminated(a, ec, at)case Create(failure)                                              => create(failure)case Watch(watchee, watcher)                                      => addWatcher(watchee, watcher)case Unwatch(watchee, watcher)                                    => remWatcher(watchee, watcher)case Recreate(cause)                                              => faultRecreate(cause)case Suspend()                                                    => faultSuspend()case Resume(inRespToFailure)                                      => faultResume(inRespToFailure)case Terminate()                                                  => terminate()case Supervise(child, async)                                      => supervise(child, async)case NoMessage                                                    => // only here to suppress warning}} catch handleNonFatalOrInterruptedException { e =>handleInvokeFailure(Nil, e)}val newState = calculateState// As each state accepts a strict subset of another state, it is enough to unstash if we "walk up" the state// chainval todo = if (newState < currentState) rest.reversePrepend(unstashAll()) else restif (isTerminated) sendAllToDeadLetters(todo)else if (todo.nonEmpty) invokeAll(todo, newState)
}

在这里，之前封装的Failed事件将会在handleFailure()函数中相应进行处理。

在handleFailure()函数中将会根据之前定义的监管策略来对此处取得的异常进行相应的处理。

final protected def handleFailure(f: Failed): Unit = {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 =>if (!actor.supervisorStrategy.handleFailure(this, f.child, f.cause, stats, getAllChildStats)) throw f.causecase 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))}
}

在这里，当配置的应对策略是Escalate的时候，将会直接false,导致会直接将这个异常抛到更加高一层的actor中进行处理。

akka actor父子监管的实现相关推荐

【Akka】Akka Actor生命周期
1.概述转载:Akka Actor生命周期用于自我学习. Actor的生命周期是使用Hooks体现和控制的,我们可以重写相关的Hooks,从而实现对Actor生命周期各环节的细粒度控制.各事件发生 ...
java akka actor,【Akka】Actor引用
Actor系统的实体在Actor系统中,actor之间具有树形的监管结构,并且actor可以跨多个网络节点进行透明通信. 对于一个Actor而言,其源码中存在Actor,ActorContext,A ...
Akka Actor模型的简介与Actor的创建方式
Akka Actor其具有以下特点: 系统中的所有事物都可以扮演一个Actor Actor之间完全独立在收到消息时,Actor所采取的所有动作都是并行的,在一个方法中的动作没有明确的顺序 Actor ...
akka actor java_Akka:使用非默认构造函数在Scala中定义一个actor并从Java代码创建它 - java...
Akka Scala演员必须扩展akka.actor.Actor Akka Java actor必须扩展akka.actor.UntypedActor 因此,在使用非默认构造函数定义Scala act ...
scala中akka actor例子
抛开复杂的业务逻辑,让我们从一个超级简单的例子学习Akka Actor的用法. Scala cookbook的作者Alvin Alexander在他的网站上提供了两个例子. 本文翻译.整理于他的两篇文 ...
AKKA Actor创建
Actor 类定义 Actor 类需要继承AbstractActor类实现createReceive方法,绑定各类actor收到不同类型消息对应处理不同业务逻辑默认提供了ReceiveBuilde ...
akka actor java_现代化的Java（三）——从Hello Akka说起
接下来打算花一些篇幅介绍一些异步编程的范式,也算是给自己一个学习笔记吧. 异步编程是个很诡异的领域,每个人都在谈论它,但是工作中很少有人能驾驭:很多很新潮很热的异步编程概念,运用起来却完全达不到它宣称 ...
java akka actor性能_akka-http - 如何通过Akka HTTP（Java）与Akka Actor进行交互 - 堆栈内存溢出...
话题我想通过Akka HTTP与Akka Actor进行交互 . 这个想法是要有一个系统,其中HTTP客户端调用Akka HTTP服务器方法,该方法处理对Akka Actor的请求. actor处理 ...
Akka Actor Inbox_信箱
2019独角兽企业重金招聘Python工程师标准>>> Akka Actor Inbox_信箱 Inbox_send_receive Inbox 形象的表示为Actor的信箱,具有收 ...

akka actor父子监管的实现

akka actor父子监管的实现相关推荐

最新文章

热门文章