Akka并发编程——第五节:Actor模型(四)

  1. 云栖社区>
  2. 博客>
  3. 正文

Akka并发编程——第五节:Actor模型(四)

周志湖 2016-04-02 22:24:00 浏览2078
展开阅读全文

本节主要内容:
1. 停止Actor

1. 停止Actor

(1)通过ActorSystem.shutdown方法停止所有 Actor的运行

/*
*停止Actor:ActorSystem.shutdown方法
*/
object Example_10 extends App{
  import akka.actor.Actor
  import akka.actor.ActorSystem
  import akka.actor.Props

  class FirstActor extends Actor with ActorLogging{

    var child:ActorRef = context.actorOf(Props[MyActor], name = "myActor")
    def receive = {
      //向MyActor发送消息
      case x => child ! x;log.info("received "+x)
    }
    override def postStop(): Unit = {
      log.info("postStop In FirstActor")
    }
  }

  class MyActor extends Actor with ActorLogging{
    def receive = {
      case "test" => log.info("received test");
      case _      => log.info("received unknown message");
    }
    override def postStop(): Unit = {
      log.info("postStop In MyActor")
    }
  }
  val system = ActorSystem("MyActorSystem")
  val systemLog=system.log

  //创建FirstActor对象
  val firstactor = system.actorOf(Props[FirstActor], name = "firstActor")

  systemLog.info("准备向firstactor发送消息")
  //向firstactor发送消息
  firstactor!"test"
  firstactor! 123
  //关闭ActorSystem,停止所有Acotr运行
  system.shutdown()
}

代码运行结果:

[INFO] [04/02/2016 21:51:34.440] [main] [ActorSystem(MyActorSystem)] 准备向firstactor发送消息
[INFO] [04/02/2016 21:51:34.441] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myActor] received test
[INFO] [04/02/2016 21:51:34.441] [MyActorSystem-akka.actor.default-dispatcher-2] [akka://MyActorSystem/user/firstActor] received test
[INFO] [04/02/2016 21:51:34.441] [MyActorSystem-akka.actor.default-dispatcher-2] [akka://MyActorSystem/user/firstActor] received 123
[INFO] [04/02/2016 21:51:34.441] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myActor] received unknown message
[INFO] [04/02/2016 21:51:34.446] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myActor] postStop In MyActor
[INFO] [04/02/2016 21:51:34.447] [MyActorSystem-akka.actor.default-dispatcher-5] [akka://MyActorSystem/user/firstActor] postStop In FirstActor

(2)通过context.stop方法停止Actor的运行

/*
*停止Actor:context.stop方法
*/
object Example_11 extends App{
  import akka.actor.Actor
  import akka.actor.ActorSystem
  import akka.actor.Props

  class FirstActor extends Actor with ActorLogging{

    var child:ActorRef = context.actorOf(Props[MyActor], name = "myActor")
    def receive = {
      case "stop"=>context.stop(child)
      case x =>{
        //向MyActor发送消息
        child ! x
        log.info("received "+x)
      }

    }
    override def postStop(): Unit = {
      log.info("postStop In FirstActor")
    }
  }

  class MyActor extends Actor with ActorLogging{
    def receive = {
      case "test" => log.info("received test");
      case _      => log.info("received unknown message");
    }
    override def postStop(): Unit = {
      log.info("postStop In MyActor")
    }
  }
  val system = ActorSystem("MyActorSystem")
  val systemLog=system.log

  //创建FirstActor对象
  val firstactor = system.actorOf(Props[FirstActor], name = "firstActor")

  systemLog.info("准备向firstactor发送消息")
  //向firstactor发送消息
  firstactor!"test"
  firstactor! 123
  firstactor!"stop"

}

代码运行结果:

[INFO] [04/02/2016 22:02:48.760] [main] [ActorSystem(MyActorSystem)] 准备向firstactor发送消息
[INFO] [04/02/2016 22:02:48.761] [MyActorSystem-akka.actor.default-dispatcher-5] [akka://MyActorSystem/user/firstActor/myActor] received test
[INFO] [04/02/2016 22:02:48.761] [MyActorSystem-akka.actor.default-dispatcher-2] [akka://MyActorSystem/user/firstActor] received test
[INFO] [04/02/2016 22:02:48.762] [MyActorSystem-akka.actor.default-dispatcher-2] [akka://MyActorSystem/user/firstActor] received 123
[INFO] [04/02/2016 22:02:48.762] [MyActorSystem-akka.actor.default-dispatcher-5] [akka://MyActorSystem/user/firstActor/myActor] received unknown message
[INFO] [04/02/2016 22:02:48.763] [MyActorSystem-akka.actor.default-dispatcher-5] [akka://MyActorSystem/user/firstActor/myActor] postStop In MyActor

代码的重点为

class FirstActor extends Actor with ActorLogging{

    var child:ActorRef = context.actorOf(Props[MyActor], name = "myActor")
    def receive = {
      case "stop"=>context.stop(child)
      case x =>{
        //向MyActor发送消息
        child ! x
        log.info("received "+x)
      }

    }
    override def postStop(): Unit = {
      log.info("postStop In FirstActor")
    }
  }

中的case “stop”=>context.stop(child),直接通过context.stop方法停止Actor的运行。注意程序中并没有使用system.shutdown方法,因此整个程序的不会停止,如下图所示
这里写图片描述

(3)通过akka.actor.PoisonPill消息停止Actor的运行

/*
*停止Actor:使用akka.actor.PoisonPill
*/
object Example_12 extends App{
  import akka.actor.Actor
  import akka.actor.ActorSystem
  import akka.actor.Props
  import akka.actor.PoisonPill

  class FirstActor extends Actor with ActorLogging{

    var child:ActorRef = context.actorOf(Props[MyActor], name = "myActor")
    def receive = {
      //向child发送PoisonPill停止其运行
      case "stop"=>child!PoisonPill
      case x =>{
        //向MyActor发送消息
        child ! x
        log.info("received "+x)
      }

    }
    override def postStop(): Unit = {
      log.info("postStop In FirstActor")
    }
  }

  class MyActor extends Actor with ActorLogging{
    def receive = {
      case "test" => log.info("received test");
      case _      => log.info("received unknown message");
    }
    override def postStop(): Unit = {
      log.info("postStop In MyActor")
    }
  }
  val system = ActorSystem("MyActorSystem")
  val systemLog=system.log

  //创建FirstActor对象
  val firstactor = system.actorOf(Props[FirstActor], name = "firstActor")

  systemLog.info("准备向firstactor发送消息")
  //向firstactor发送消息
  firstactor!"test"
  firstactor! 123
  firstactor!"stop"

}

代码运行结果:

[INFO] [04/02/2016 22:12:09.947] [main] [ActorSystem(MyActorSystem)] 准备向firstactor发送消息
[INFO] [04/02/2016 22:12:09.947] [MyActorSystem-akka.actor.default-dispatcher-2] [akka://MyActorSystem/user/firstActor/myActor] received test
[INFO] [04/02/2016 22:12:09.947] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor] received test
[INFO] [04/02/2016 22:12:09.947] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor] received 123
[INFO] [04/02/2016 22:12:09.947] [MyActorSystem-akka.actor.default-dispatcher-2] [akka://MyActorSystem/user/firstActor/myActor] received unknown message
[INFO] [04/02/2016 22:12:09.951] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myActor] postStop In MyActor

代码与Exampel_11中的不同之处在于

//向child发送PoisonPill停止其运行
case "stop"=>child!PoisonPill

它使用不是context.stop方法,而是向MyActor发送了PoisonPill消息,其它代码不变。

还有一种gracefulStop方法可以停止Actor的运行,这部分内容等了解完Future类的作用原理之后再来讨论



Scala学习(公众微信号:ScalaLearning)每天为大家带来一点Scala语言、Spark、Kafka、Flink、AKKA等大数据技术干货及相关技术资讯。技术永无止境,勇攀高峰,一往直前!
觉得文章不错?扫描关注

这里写图片描述

网友评论

登录后评论
0/500
评论
周志湖
+ 关注