akka学习教程(四) actor生命周期

Bertha 。 2022-07-13 10:24 228阅读 0赞

# akka系列文章目录 #

*  [akka学习教程(十四) akka分布式实战][akka_ akka]
 *  [akka学习教程(十三) akka分布式][akka_ akka 1]
 *  [akka学习教程(十二) Spring与Akka的集成][akka_ Spring_Akka]
 *  [akka学习教程(十一) akka持久化][akka_ akka 2]
 *  [akka学习教程(十) agent][akka_ agent]
 *  [akka学习教程(九) STM软件事务内存][akka_ STM]
 *  [akka学习教程(八) Actor中的Future-询问模式][akka_ Actor_Future-]
 *  [akka学习教程(七) 内置状态转换Procedure][akka_ _Procedure]
 *  [akka学习教程(六) 路由器Router][akka_ _Router]
 *  [akka学习教程(五) inbox消息收件箱][akka_ inbox]
 *  [akka学习教程(四) actor生命周期][akka_ actor]
 *  [akka学习教程(三) 不可变对象][akka_]
 *  [akka学习教程(二)HelloWord][akka_HelloWord]
 *  [akka学习教程(一)简介][akka]

![这里写图片描述][SouthEast]

import akka.actor.UntypedActor;
    import akka.event.Logging;
    import akka.event.LoggingAdapter;
    
    public class MyWork extends UntypedActor { 
    
        LoggingAdapter logger = Logging.getLogger(getContext().system(), this);
    
        public static enum Msg{
            WORKING, DONE, CLOSE;
        }
    
        /** * * / @Override public void preStart() { logger.info("myWork starting."); } @Override public void postStop() throws Exception { logger.info("myWork stoping.."); } @Override public void onReceive(Object msg) { try { if(msg == Msg.WORKING){ logger.info("i am working"); }else if(msg == Msg.DONE){ logger.info("stop working"); }else if(msg == Msg.CLOSE){ logger.info("stop close"); getSender().tell(Msg.CLOSE, getSelf()); getContext().stop(getSelf()); }else { unhandled(msg); } }catch (Exception e){ e.printStackTrace(); } } }

import akka.actor.ActorRef;
    import akka.actor.Terminated;
    import akka.actor.UntypedActor;
    import akka.event.Logging;
    import akka.event.LoggingAdapter;
    
    public class WatchActor extends UntypedActor { 
    
      LoggingAdapter logger = Logging.getLogger(getContext().system(), this);
    
      /** * 监听一个actor * @param actorRef */
      public WatchActor(ActorRef actorRef){
        getContext().watch(actorRef);
      }
    
      @Override
      public void onReceive(Object msg) throws InterruptedException {
        if(msg instanceof Terminated){
          //这里简单打印一下，然后停止system
          logger.error(((Terminated)msg).getActor().path() + " has Terminated. now shutdown the system");
          getContext().system().shutdown();
        }else{
          unhandled(msg);
        }
    
      }
    
    }

import akka.actor.ActorRef;
    import akka.actor.ActorSystem;
    import akka.actor.PoisonPill;
    import akka.actor.Props;
    import com.typesafe.config.ConfigFactory;
    
    public class Main { 
    
      public static void main(String[] args) {
        //创建ActorSystem。一般来说，一个系统只需要一个ActorSystem。
        //参数1：系统名称。参数2：配置文件
        ActorSystem system = ActorSystem.create("Hello", ConfigFactory.load("akka.config"));
        ActorRef myWork = system.actorOf(Props.create(MyWork.class), "MyWork");
        ActorRef watchActor = system.actorOf(Props.create(WatchActor.class, myWork), "WatchActor");
    
        myWork.tell(MyWork.Msg.WORKING, ActorRef.noSender());
        myWork.tell(MyWork.Msg.DONE, ActorRef.noSender());
    
        //中断myWork
        myWork.tell(PoisonPill.getInstance(), ActorRef.noSender());
      }
    }

输出结果：

[INFO] [01/05/2017 15:26:15.705] [Hello-akka.actor.default-dispatcher-4] [akka://Hello/user/MyWork] myWork starting.
    [INFO] [01/05/2017 15:26:15.730] [Hello-akka.actor.default-dispatcher-4] [akka://Hello/user/MyWork] i am working
    [INFO] [01/05/2017 15:26:15.730] [Hello-akka.actor.default-dispatcher-4] [akka://Hello/user/MyWork] stop working
    [INFO] [01/05/2017 15:26:15.736] [Hello-akka.actor.default-dispatcher-4] [akka://Hello/user/MyWork] myWork stoping..
    [ERROR] [01/05/2017 15:26:15.749] [Hello-akka.actor.default-dispatcher-10] [akka://Hello/user/WatchActor] akka://Hello/user/MyWork has Terminated. now shutdown the system

# 监督 #

这里可以对actor进行监督，在actor报出异常的时候进行处理。由于这个和生命周期也有关，所以放到这里一起说了。

akka监督策略有两种：

*  OneForOneStrategy  
    
    
     *  只对出问题的子actor进行处理. 这是默认策略
 *  AllForOneStrategy  
    
    
     *  对子actor以及他的所有兄弟actor进行处理

定义监督者的行为：

package akka.strategy;
    
    import akka.actor.OneForOneStrategy;
    import akka.actor.Props;
    import akka.actor.SupervisorStrategy;
    import akka.actor.UntypedActor;
    import akka.japi.Function;
    import scala.concurrent.duration.Duration;
    
    import java.util.concurrent.TimeUnit;
    
    /** * Created by liubenlong on 2017/1/9. * 监督者，监督策略 */
    public class SuperVisor extends UntypedActor{ 
    
        /** * 配置自己的strategy * @return */
        @Override
        public SupervisorStrategy supervisorStrategy(){
            return new OneForOneStrategy(3, Duration.create(1, TimeUnit.MINUTES),//一分钟内重试3次，超过则kill掉actor
                    new Function<Throwable, SupervisorStrategy.Directive>() {
                        @Override
                        public SupervisorStrategy.Directive apply(Throwable throwable) throws Exception {
                            if(throwable instanceof ArithmeticException){
       //ArithmeticException是出现异常的运算条件时，抛出此异常。例如，一个整数“除以零”时，抛出此类的一个实例。
                                System.out.println("meet ArithmeticException ,just resume.");
                                return  SupervisorStrategy.resume();//继续; 重新开始; 恢复职位;
                            }else if(throwable instanceof NullPointerException){
                                System.out.println("meet NullPointerException , restart.");
                                return SupervisorStrategy.restart();
                            }else if(throwable instanceof IllegalArgumentException){
                                System.out.println("meet IllegalArgumentException ,stop.");
                                return SupervisorStrategy.stop();
                            }else{
                                System.out.println("escalate.");
                                return SupervisorStrategy.escalate();//使逐步升级; 使逐步上升; 乘自动梯上升;也就是交给更上层的actor处理。抛出异常
                            }
                        }
                    });
        }
    
        @Override
        public void onReceive(Object o) throws Throwable {
            if(o instanceof Props){
                getContext().actorOf((Props)o , "restartActor");
            }else{
                unhandled(o);
            }
        }
    }

定义restartActor的实现：

package akka.strategy;
    
    import akka.actor.UntypedActor;
    import scala.Option;
    
    /** * Created by liubenlong on 2017/1/12. */
    public class RestartActor extends UntypedActor { 
    
        public  enum  Msg{
            DONE, RESTART;
        }
    
    
        @Override
        public void preStart() throws Exception {
            System.out.println("preStart hashCode=" + this.hashCode());
        }
        @Override
        public void postStop() throws Exception {
            System.out.println("postStop hashCode=" + this.hashCode());
        }
    
    
    
        @Override
        public void preRestart(Throwable reason, Option<Object> message) throws Exception {
            System.out.println("preRestart hashCode=" + this.hashCode());
        }
        @Override
        public void postRestart(Throwable reason) throws Exception {
            super.postRestart(reason);
            System.out.println("postRestart hashCode=" + this.hashCode());
        }
    
    
    
        @Override
        public void onReceive(Object o) throws Throwable {
            if(o == Msg.DONE){
                getContext().stop(getSelf());
            }else if(o == Msg.RESTART){
                System.out.println(((Object) null).toString());
                //抛出异常，默认会被restart，但这里会resume
                //double a = 1/0;
            }else{
                unhandled(o);
            }
    
        }
    }

定义main方法：

package akka.strategy;
    
    import akka.actor.*;
    import com.typesafe.config.ConfigFactory;
    
    /** * Created by liubenlong on 2017/1/12. */
    public class Main { 
    
        public static void main(String[] args) {
            ActorSystem system = ActorSystem.create("strategy", ConfigFactory.load("akka.config"));
            ActorRef superVisor = system.actorOf(Props.create(SuperVisor.class), "SuperVisor");
            superVisor.tell(Props.create(RestartActor.class), ActorRef.noSender());
    
            ActorSelection actorSelection = system.actorSelection("akka://strategy/user/SuperVisor/restartActor");//这是akka的路径。restartActor是在SuperVisor中创建的。
    
            for(int i = 0 ; i < 100 ; i ++){
                actorSelection.tell(RestartActor.Msg.RESTART, ActorRef.noSender());
            }
        }
    
    }

输出结果（省略部分异常）：

com.intellij.rt.execution.application.AppMain akka.strategy.Main
    preStart    hashCode=463293128
    meet NullPointerException , restart.
    preRestart    hashCode=463293128
    preStart    hashCode=1195770342
    postRestart    hashCode=1195770342
    meet NullPointerException , restart.
    preRestart    hashCode=1195770342
    preStart    hashCode=1139511720
    postRestart    hashCode=1139511720
    meet NullPointerException , restart.
    preRestart    hashCode=1139511720
    preStart    hashCode=1906622979
    postRestart    hashCode=1906622979
    meet NullPointerException , restart.
    postStop    hashCode=1906622979
    [ERROR] [01/12/2017 10:52:05.563] [strategy-akka.actor.default-dispatcher-2] [akka://strategy/user/SuperVisor/restartActor] null
    java.lang.NullPointerException
        at akka.strategy.RestartActor.onReceive(RestartActor.java:44)
        at akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:165)

结果分析：  
preStart是`restartActor`初始化时调用的，他的hashcode是`463293128`，接着遇到空指针异常，根据自定义策略，会重启改actor。此时会调用`preRestart`，注意他的hashcode依然是`463293128`，因为`preRestart`是在正式启动前在老的actor上调用的。  
随后打印出`preStart`,说明新的actor开始创建了，他的hashcode是`1195770342`,新的actor实例将替代旧的实例工作，**这说明同一个restartActor工作的过程中，未必真的是同一个actor**。重启完成之后调用`postRestart`.

再经历3次（自定义策略配置的）重启以后，达到重启上限，系统将直接关闭该actor。

## 选择actor ##

上面程序中使用到了`ActorSelection actorSelection = system.actorSelection("akka://strategy/user/SuperVisor/restartActor");`进行actor的选择。  
工作过程中可能会存在成千上万的actor，可以通过`actorSelection`方便的选择actor进行消息投递，其支持通配符匹配`getContext().actorSelection("/user/worker_*")`.

# 参考资料 #

*  书籍《java高并发程序设计》
 *  [AKKA官方文档][AKKA]

[akka_ akka]: http://blog.csdn.net/liubenlong007/article/details/54603296
[akka_ akka 1]: http://blog.csdn.net/liubenlong007/article/details/54601998
[akka_ Spring_Akka]: http://blog.csdn.net/liubenlong007/article/details/54601448
[akka_ akka 2]: http://blog.csdn.net/liubenlong007/article/details/54584303
[akka_ agent]: http://blog.csdn.net/liubenlong007/article/details/54576611
[akka_ STM]: http://blog.csdn.net/liubenlong007/article/details/54576412
[akka_ Actor_Future-]: http://blog.csdn.net/liubenlong007/article/details/54575312
[akka_ _Procedure]: http://blog.csdn.net/liubenlong007/article/details/54574878
[akka_ _Router]: http://blog.csdn.net/liubenlong007/article/details/54574064
[akka_ inbox]: http://blog.csdn.net/liubenlong007/article/details/54377820
[akka_ actor]: http://blog.csdn.net/liubenlong007/article/details/54093889
[akka_]: http://blog.csdn.net/liubenlong007/article/details/54093794
[akka_HelloWord]: http://blog.csdn.net/liubenlong007/article/details/53782971
[akka]: http://blog.csdn.net/liubenlong007/article/details/53782966
[SouthEast]: /images/20220713/645a36ca96554bd08ab8d5bcde7d99a6.png
[AKKA]: http://akka.io/docs/