22、聊聊akka(二)监控和监视

雨点打透心脏的1/2处 2022-05-27 09:18 264阅读 0赞

虽然通过充分利用多核CPU的计算能力把数据处理运算进行并行处理，提高系统整体效率，对现今大数据普遍盛行的系统计算要求还是远远不足的，只有通过硬件平行拓展（scale-out）形成机群并在之上实现分布式运算才能正真符合新环境对软件程序的要求。

**Akka程序是由多个Actor组成的**。它的工作原理是把一项大运算分割成许多小任务然后把这些任务托付给多个Actor去运算。Actor不单可以在当前JVM中运行，也可以跨JVM在任何机器上运行，这基本上就是Akka程序实现分布式运算的关键了。当然，这也有赖于Akka提供的包括监管、监视各种Actor角色，各式运算管理策略和方式包括容错机制、内置线程管理、远程运行（remoting）等，以及一套分布式的消息系统来协调、控制整体运算的安全进行。

Actor是Akka系统中的最小运算单元。每个Actor只容许单一线程，这样来说Actor就是一种更细小单位的线程。Akka的编程模式和其内置的线程管理功能使用户能比较自然地实现多线程并发编程。Actor的主要功能就是在单一线程里运算维护它的内部状态，那么它的内部状态肯定是可变的（mutable state），但因为每个Actor都是独立的单一线程运算单元，加上运算是消息驱动的（message-driven），只容许线性流程，Actor之间运算结果互不影响，所以从Akka整体上来讲Actor又好像是纯函数不可变性的（pure immutable）。

**消息驱动模式的好处是可以实现高度的松散耦合（loosely-coupling）**,因为系统部件之间不用软件接口，而是通过消息来进行系统集成的。消息驱动模式支持了每个Actor的独立运算环境，又可以在运行时按需要灵活的对系统Actor进行增减，伸缩自如，甚至可以在运行时（runtime）对系统部署进行调配。Akka的这些鲜明的特点都是通过消息驱动来实现的。

## 容错 ##

将错误（崩溃）孤立出来，不会导致整个系统崩溃（隔离故障组件），备份组件可以替换崩溃组件（冗余）（可恢复性）  
容错方式：Restart, Resume, Stop, Escalate  
崩溃原因：网络，第三方服务，硬件故障  
Akka容错：分离业务逻辑（receive）和容错逻辑(supervisorStrategy)  
父actor自动成为子actor的supervisor  
supervisor不fix子actor，而是简单的呈现如何恢复的一个判断==>  
List(Restart, //重启并替换原actor,mailbox消息可继续发送，  
//但是接收会暂停至替换完成，重启默认重启所有子actor  
Resume, //同一个actor不重启，忽略崩溃，继续处理下一个消息  
Stop, //terminated 不再处理任何消息，剩余消息会进入死信信箱  
Escalate//交给上层处理

Akka的Actor组织是一个层级结构。下层Actor是由直接上一层Actor产生，形成一种父子Actor关系。父级Actor除维护自身状态之外还必须负责处理下一层子级Actor所发生的异常，形成一种树形父子层级监管结构。任何子级Actor在运算中发生异常后立即将自己和自己的子级Actor运算挂起，并将下一步行动交付给自己的父级Actor决定。父级Actor对发生异常的子级Actor有以下几种处理方式：

> 1、恢复运算（Resume）：不必理会异常，保留当前状态，跳过当前异常消息，照常继续处理其它消息  
> 2、重新启动（Restart）：清除当前状态，保留邮箱及内容，终止当前Actor，再重新构建一个新的Actor实例，沿用原来的消息地址ActorRef继续工作  
> 3、彻底终止（Stop）：销毁当前Actor及ActorRef邮箱，把所有消息导向DeadLetter队列。  
> 4、向上提交（Esculate）：如果父级无法处理子级异常，则这种情况也视为父级出现的异常。按照规定，父级会将自己和子级Actor运算暂停挂起并把子级Actor实际产生的异常当作自己发生的异常提交给上一层父级处理（也就是说异常信息的发送者sender变成了父级Actor）。

Akka处理异常的方式简单直接：如果发生异常就先暂停挂起然后交给直属父级Actor去处理。这就把异常封闭在这个Actor的监管链条里。Akka系统的监管链条实际代表一个功能的分散封闭运算，所以一个监管链条里发生的异常不会影响其它监管链条。换句话说就是Actor发生异常是封闭在它所属的功能内部的，一个功能发生异常不会影响其它功能。而在行令式程序中，如果没有try-catch，任何一段产生异常的代码都会导致整个程序中断。

Akka提供了OneForOneStrategy和AllForOneStrategy两种对待异常Actor的策略配置，策略中定义了对下属子级发生的各种异常的处理方式。异常处理策略是以策略施用对象分类的，如下：

OneForOneStrategy：只针对发生异常的Actor施用策略  
AllForOneStrategy：虽然一个直属子级Actor发生了异常，监管父级Actor把它当作所有下属子级同时发生了相同异常，对所有子级Actor施用策略  
正常情况下你应该采用OneForOneStrategy，而且这也是Akka默认采用的机制。

Akka对待这种父子监管的原则保证了在Akka系统中不会出现任何孤儿，也就是说保证不会出现断裂的监管树。这就要求当任何一个Actor在暂停挂起前都要保证先暂停挂起它的所有直属子级Actor，而子级则必须先暂停挂起它们的直属子级，如此递归。同样，任何Actor在重启（Restart）时也必须递归式地重启直属子级，因为重启一个Actor需要先停止再启动，我们必须肯定在停止时不会产生孤儿Actor。如果一个父级Actor无法处理子级异常需要向上提交（Esculate）的话，首先它需要采取递归方式来暂停挂起自身以下的监管链条。它的直属父级Actor会按自己的异常处理策略来对待提交上来的异常，处理的结果将会递归式沿着监管树影响属下的所有子子孙孙。但如果这个级别的Actor异常处理策略还是无法覆盖这个异常时，它又会挂起自己，再向上提交（Esculate）。那么如果到达了顶级Actor又如何向上提交呢？Akka系统最终的异常处理策略可以在config文件里配置：

# The guardian "/user" will use this class to obtain its supervisorStrategy.
    # It needs to be a subclass of akka.actor.SupervisorStrategyConfigurator.
    # In addition to the default there is akka.actor.StoppingSupervisorStrategy.
        guardian-supervisor-strategy = "akka.actor.DefaultSupervisorStrategy"

默认策略是DefaultSupervisorStrategy。以下是Akka提供的默认策略：

final val defaultDecider: Decider = {
        case _: ActorInitializationException ⇒ Stop
        case _: ActorKilledException         ⇒ Stop
        case _: DeathPactException           ⇒ Stop
        case _: Exception                    ⇒ Restart
      }
    final val defaultStrategy: SupervisorStrategy = {
        OneForOneStrategy()(defaultDecider)
      }

前面三种异常直属父级直接终止子级Actor，其它类型重启。当然我们可以在这个默认策略之上再添加自定义的一些异常处理策略

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

Akka绝对不容许有孤儿Actor存在（断裂的监管树），所以停止任何一个Actor，它下属的子子孙孙都会自下而上依次停止运算。为了更好的理解Actor的监管策略，我们必须先从了解Actor的生命周期（lift-cycle）开始。一个Actor从构建产生ActorRef开始到彻底终止为整个生命周期。其中可以发生多次重启（Restart）。我们在下面对Actor的开始、终止、重启这三个环节中发生的事件进行描述：  
1、开始

@Override
    public void preStart(){
        initDB
    }

2、终止

@Override
    public void postStop(){
        db.release
    }

3、重启  
重启是Actor生命周期里一个最重要的环节。在一个Actor的生命周期里可能因为多种原因发生重启（Restart）。造成一个Actor需要重启的原因可能有下面几个：

1、在处理某特定消息时造成了系统性的异常，必须通过重启来清理系统错误  
2、内部状态毁坏，必须通过重启来重新构建状态  
3、在处理消息时无法使用到一些依赖资源，需要重启来重新配置资源

重启是一个先停止再开始的过程。父级Actor通过递归方式先停止下面的子孙Actor，那么在启动过程中这些停止的子孙Actor是否会自动构建呢？这里需要特别注意：因为父级Actor是通过Props重新构建的，如果子级Actor的构建是在父级Actor的类构建器内而不是在消息处理函数内构建的，那么子级Actor会自动构建。**Akka提供了preRestart和postRestart两个事件接口**。preRestart发生在停止之前，postRestart发生在开始前，如下：

@Override
    public void preRestart(Throwable reason, Option<Object> message) 
    
    @Override
    public void postRestart(Throwable reason)

很多时候由于外界原因，Actor的重启无法保证一次成功。这种现象在使用依赖资源如数据库、网络连接等最为明显。我们前面介绍过的异常处理策略中就包含了重试（retry）次数及最长重试时间。

case class OneForOneStrategy(
      maxNrOfRetries:              Int      = -1,
      withinTimeRange:             Duration = Duration.Inf,
      override val loggingEnabled: Boolean  = true)(val decider: SupervisorStrategy.Decider)
      extends SupervisorStrategy {
       ...}

kka提供了context.watch和context.unwatch来设置通过ActorRef对任何Actor的终止状态监视，无须父子级别关系要求。下面是Akka提供的这两个函数：

def watch(actorRef: ActorRef): Unit = {
        watching += actorRef
        actorRef.asInstanceOf[InternalActorRef].sendSystemMessage(Watch(actorRef.asInstanceOf[InternalActorRef], this))
      }
    
      def unwatch(actorRef: ActorRef): Unit = {
        watching -= actorRef
        actorRef.asInstanceOf[InternalActorRef].sendSystemMessage(Unwatch(actorRef.asInstanceOf[InternalActorRef], this))
      }

worker java.lang.NullPointerException: run 
    [INFO] [04/16/2018 10:09:01.153] [FaultHandlingTest-akka.actor.default-dispatcher-4] [akka://FaultHandlingTest/user/supervisor/child] Worker starting.
    meet NullPointerException , restart.
    preRestart 0   hashCode=1876167812
    [WARN] [04/16/2018 10:09:01.155] [FaultHandlingTest-akka.actor.default-dispatcher-2] [akka.tcp://FaultHandlingTest@127.0.0.1:2551/system/cluster/core/daemon/downingProvider] Don't use auto-down feature of Akka Cluster in production. See 'Auto-downing (DO NOT USE)' section of Akka Cluster documentation.
    postRestart 1  hashCode=801569476
    [ERROR] [04/16/2018 10:09:01.159] [FaultHandlingTest-akka.actor.default-dispatcher-15] [akka://FaultHandlingTest/user/supervisor/child] run 
    java.lang.NullPointerException: run 
        at org.akka.faulttolerance.Listener.onReceive(Listener.java:44)
    [INFO] [04/16/2018 10:09:01.159] [FaultHandlingTest-akka.actor.default-dispatcher-15] [akka://FaultHandlingTest/user/supervisor/child] Worker stoping..
    [INFO] [04/16/2018 10:09:01.183] [FaultHandlingTest-akka.actor.default-dispatcher-15] [akka://FaultHandlingTest/user/supervisor/child] Worker starting.

监督者面对其下属的失败，有不同的策略。不过大致可以分为已下的四类：

1）恢复下级Actor，并且保持下级Actor的内部状态(略过处理)  
2）重新启动下级Actor，并且**清除下级Actor的内部状态** /重启并替换原actor,mailbox消息可继续发送，不能处理异常信息  
3）永久的停止下级Actor  
4）将错误逐层上传，从而暂停自己（有点像java中的throw exception）

其中需要知道的是，Akka中的每一个Actor都在监督树中，他们既可以扮演监督者的角色，也可以扮演被监督者的角色。上级Actor的状态直接影响着下级Actor的状态，因此对上面的前三条策略可以做如下补充（诠释）

1）当恢复某个Actor的时候同时也要恢复它的所有下级Actor  
2）重新启动某个Actor的时候也要重启它所有的下级Actor  
3）停止某个Actor的时候也需要停止它所有的下级Actor。

Actor的preRestart方法的默认行为就是：在这个Actor重启前，先终止它所有的下级Actor，这个过程其实就是一个递归的过程。但是这个方法是可以重写的，因此在重写的时候需要谨慎。

// 定义监督策略：
        private static SupervisorStrategy strategy = new OneForOneStrategy(-1,
                Duration.Inf(), new Function<Throwable, Directive>() {  //AllForOneStrategy
            @Override
            public Directive apply(Throwable t) {
                if (t instanceof Exception) {
                    return SupervisorStrategy.restart();
                    //return SupervisorStrategy.resume();
                } else {
                    return escalate();
                }
            }
        });
    
        @Override
        public void preStart() throws Exception {
            for (int i = 0; i < 10; i++) {
                all[i] = getContext().actorOf(Actor2.props());
                getContext().watch(all[i]);
            }
            getSelf().tell(MSG.START, getSelf());
        }
        @Override
        public void onReceive(Object message) throws Exception {
            System.out.println(name + "_" + index + " receive message" + message.toString());
            if (message instanceof MSG) {
                if (message.equals(MSG.START)) {
                    System.out.println("****START****");
                    all[(int) (Math.random() * 10)].tell(MSG.HI + ""+ count, getSelf());
                    self().tell(MSG.GOON, self());
                } else if (message.equals(MSG.GOON)){
                    all[(int) (Math.random() * 10)].tell(MSG.HI + ""+ count, getSelf());
                    self().tell(MSG.GOON, self());
                    if (++count == 100) {
                        throw new ActorKilledException("Teminate all actor!");
                    }
                } else {
                    unhandled(message);
                }
            } else if (message instanceof Terminated) {
                System.out.println(getSender() + " terminate");
            } else {
                unhandled(message);
            }
    
        }

参考：[http://www.cnblogs.com/tiger-xc/p/6760658.html][http_www.cnblogs.com_tiger-xc_p_6760658.html]  
[https://github.com/akka][https_github.com_akka]

[http_www.cnblogs.com_tiger-xc_p_6760658.html]: http://www.cnblogs.com/tiger-xc/p/6760658.html
[https_github.com_akka]: https://github.com/akka