码迷,mamicode.com
首页 > 其他好文 > 详细

Akka之BackoffSupervisor

时间:2018-10-14 14:13:45      阅读:196      评论:0      收藏:0      [点我收藏+]

标签:dex   tde   enc   异常处理   owa   strong   art   any   received   

一、背景

最近在开发一个项目,项目的各模块之间是使用akka grpc传输音频帧的,并且各模块中的actor分别都进行了persist。本周在开发过程中遇到了一个bug,就是音频帧在通行一段时间后,整个系统处于卡死状态,没有了反应。刚开始怀疑是akka grpc通信时,流中断了,或者没有传输过来,可是通过抓包和日志,发现流的每一帧已经接受到了。最后定位到问题时persist的原因,每次卡死之间都可以发现persit失败了。我们去看persist方法的源码可以发现上面的注解如下:

/**
* Asynchronously persists `event`. On successful persistence, `handler` is called with the
* persisted event. It is guaranteed that no new commands will be received by a persistent actor
* between a call to `persist` and the execution of its `handler`. This also holds for
* multiple `persist` calls per received command. Internally, this is achieved by stashing new
* commands and unstashing them when the `event` has been persisted and handled. The stash used
* for that is an internal stash which doesn‘t interfere with the inherited user stash.
*
* An event `handler` may close over persistent actor state and modify it. The `sender` of a persisted
* event is the sender of the corresponding command. This means that one can reply to a command
* sender within an event `handler`.
*
* Within an event handler, applications usually update persistent actor state using persisted event
* data, notify listeners and reply to command senders.
*
* If persistence of an event fails, [[#onPersistFailure]] will be invoked and the actor will
* unconditionally be stopped. The reason that it cannot resume when persist fails is that it
* is unknown if the event was actually persisted or not, and therefore it is in an inconsistent
* state. Restarting on persistent failures will most likely fail anyway, since the journal
* is probably unavailable. It is better to stop the actor and after a back-off timeout start
* it again.
*
* @param event event to be persisted
* @param handler handler for each persisted `event`
*/
def persist[A](event: A)(handler: A ? Unit): Unit = {
internalPersist(event)(handler)
}

从注解我们可以发现,我们在使用akka的persist特性时,如果持久化失败,相应的actor就会被stop掉。因此,如果你继续往它发消息是没有任何反应。并且这个注解建议我们使用Backoff来重启这个Actor。(建议,我们在把actor持久化到本地或者使用redis等插件持久化到数据库时,最好选择持久化的方法,不然会使用java持久化,会出现WARN,因为默认的java持久化效率不是很好)。因此,我们来学习一下,这种BackOff重启的方式。

二、普通的监控和重启

我们都知道Actor之间是有父子关系的,如果子Actor出现了异常,是会进行上报,并且使用策略来进行相应的处理,其中最常用的策略就是OneForOneStrategy,也就是只针对发生异常的Actor施用策略,策略中定义了对下属子级发生的各种异常的处理方式。而默认的处理方式是这样的:

final val defaultDecider: Decider = {
case _: ActorInitializationException ? Stop
case _: ActorKilledException ? Stop
case _: DeathPactException ? Stop
case _: Exception ? Restart
}

但是,我们可以加上自己的一些特殊处理方式,例如

override val supervisorStrategy =
OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
case _: ArithmeticException => Resume
case _: MyException => Restart
case t =>
super.supervisorStrategy.decider.applyOrElse(t, (_: Any) => Escalate)
}

三、BackoffSupervisor

但是,如果我们的actor是通过system.actorOf进行启动的,我们就很难通过上述这个方式去自定义自己的异常处理方式。并且,如果我们想进行个细粒度的控制,例如在actor发生异常以后多久去处理它。这种情况我们就可以使用BackoffSupervisor去实现。

我们可以这样理解BackoffSupervisor。实际上BackoffSupervisor与定义了supervisorStrategy的Actor有所不同。我们应该把BackoffSupervisor看作是一个一体化的Actor。当然,它的实现方式还是由一对父子Actor组成。监管策略(SupervisorStrategy)是在BackoffSupervisor的内部实现的。从外表上BackoffSupervisor就像是一个Actor,运算逻辑是在子级Actor中定义的,所谓的父级Actor除监管之外没有任何其它功能,我们甚至没有地方定义父级Actor的功能,它的唯一功能是转发收到的信息给子级,是嵌入BackoffSupervisor里的。所以我们虽然发送消息给BackoffSupervisor,但实际上是在与它的子级交流。下面我们可以通过一个例子来看看怎么使用:

object InnerChild {
case class TestMessage(msg: String)
class ChildException extends Exception

def props = Props[InnerChild]
}
class InnerChild extends Actor with ActorLogging {
import InnerChild._
override def receive: Receive = {
case TestMessage(msg) => //模拟子级功能
log.info(s"Child received message: ${msg}")
}
}
object Supervisor {
def props: Props = { //在这里定义了监管策略和child Actor构建
def decider: PartialFunction[Throwable, SupervisorStrategy.Directive] = {
case _: InnerChild.ChildException => SupervisorStrategy.Restart
}

val options = Backoff.onFailure(InnerChild.props, "innerChild", 1 second, 5 seconds, 0.0)
.withManualReset
.withSupervisorStrategy(
OneForOneStrategy(maxNrOfRetries = 5, withinTimeRange = 5 seconds)(
decider.orElse(SupervisorStrategy.defaultDecider)
)
)
BackoffSupervisor.props(options)
}
}
//注意:下面是Supervisor的父级,不是InnerChild的父级
object ParentalActor {
case class SendToSupervisor(msg: InnerChild.TestMessage)
case class SendToInnerChild(msg: InnerChild.TestMessage)
case class SendToChildSelection(msg: InnerChild.TestMessage)
def props = Props[ParentalActor]
}
class ParentalActor extends Actor with ActorLogging {
import ParentalActor._
//在这里构建子级Actor supervisor
val supervisor = context.actorOf(Supervisor.props,"supervisor")
supervisor ! BackoffSupervisor.getCurrentChild //要求supervisor返回当前子级Actor
var innerChild: Option[ActorRef] = None //返回的当前子级ActorRef
val selectedChild = context.actorSelection("/user/parent/supervisor/innerChild")
override def receive: Receive = {
case BackoffSupervisor.CurrentChild(ref) => //收到子级Actor信息
innerChild = ref
case SendToSupervisor(msg) => supervisor ! msg
case SendToChildSelection(msg) => selectedChild ! msg
case SendToInnerChild(msg) => innerChild foreach(child => child ! msg)
}

}
object BackoffSupervisorDemo extends App {
import ParentalActor._
val testSystem = ActorSystem("testSystem")
val parent = testSystem.actorOf(ParentalActor.props,"parent")
Thread.sleep(1000) //wait for BackoffSupervisor.CurrentChild(ref) received
parent ! SendToSupervisor(TestMessage("Hello message 1 to supervisor"))
parent ! SendToInnerChild(TestMessage("Hello message 2 to innerChild"))
parent ! SendToChildSelection(TestMessage("Hello message 3 to selectedChild"))
scala.io.StdIn.readLine()
testSystem.terminate()
}

运行结果如下:

[INFO] [10/13/2018 16:11:48.167] [testSystem-akka.actor.default-dispatcher-2] [akka://testSystem/user/parent/supervisor/innerChild] Child received message: Hello message 1 to supervisor
[INFO] [10/13/2018 16:11:48.177] [testSystem-akka.actor.default-dispatcher-2] [akka://testSystem/user/parent/supervisor/innerChild] Child received message: Hello message 2 to innerChild
[INFO] [10/13/2018 16:11:48.179] [testSystem-akka.actor.default-dispatcher-3] [akka://testSystem/user/parent/supervisor/innerChild] Child received message: Hello message 3 to selectedChild

从结果可以看出,虽然在上面的例子里我们分别向supervisor,innerChild,selectedChild发送消息。但最终所有消息都是由InnerChild响应的。

Akka之BackoffSupervisor

标签:dex   tde   enc   异常处理   owa   strong   art   any   received   

原文地址:https://www.cnblogs.com/junjiang3/p/9785373.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!