Akka并发编程——第八节:Actor模型(七)

本节主要内容

停止运行Typed Actor

当Typed Actor不再需要时要将其停止,有3种方法停止Typed Actor的运行:

(1)通过system.shutdown()停止ActorSystem中所有的Typed Actor;

(2)调用TypedActor(system).stop(mySquarer)停止指定的Typed Actor;

(3)调用TypedActor(system).poisonPill(otherSquarer)停止指定的Typed Actor。

具体使用代码如下:

/*
 * 停止Typed Actor
 */
object Example_3 extends  App {

  import akka.event.Logging
  import scala.concurrent.{ Promise, Future }
  import akka.actor.{ TypedActor, TypedProps }
  import scala.concurrent.duration._

  trait Squarer {
    //fire-and-forget消息
    def squareDontCare(i: Int): Unit
    //非阻塞send-request-reply消息
    def square(i: Int): Future[Int]
    //阻塞式的send-request-reply消息
    def squareNowPlease(i: Int): Option[Int]
    //阻塞式的send-request-reply消息
    def squareNow(i: Int): Int
  }

  //混入PostStop和PreStart
  class SquarerImpl(val name: String) extends  Squarer with PostStop with PreStart {
    import TypedActor.context
    val log = Logging(context.system,TypedActor.self.getClass())
    def this() = this("SquarerImpl")

    def squareDontCare(i: Int): Unit = i * i
    def square(i: Int): Future[Int] = Promise.successful(i * i).future
    def squareNowPlease(i: Int): Option[Int] = Some(i * i)
    def squareNow(i: Int): Int = i * i

    def postStop(): Unit={
      log.info ("TypedActor Stopped")
    }
    def preStart(): Unit={
      log.info ("TypedActor  Started")
    }
  }

  val system = ActorSystem("TypedActorSystem")
  val log = Logging(system, this.getClass)

  //使用默认构造函数创建Typed Actor
  val mySquarer: Squarer =
    TypedActor(system).typedActorOf(TypedProps[SquarerImpl](),"mySquarer")

  //使用非默认构造函数创建Typed Actor
  val otherSquarer: Squarer =
    TypedActor(system).typedActorOf(TypedProps(classOf[Squarer],
      new SquarerImpl("SquarerImpl")), "otherSquarer")

  //Request-reply-with-future 消息发送
  val fSquare = mySquarer.square(10)
  val result = Await.result(fSquare, 5 second)
  log.info("fSquare="+result)

  //调用poisonPill方法停止Actor运行
  TypedActor(system).poisonPill(otherSquarer)

  //调用stop方法停止Actor运行
  TypedActor(system).stop(mySquarer)

  //system.shutdown()
}

代码运行结果如下所示。

[INFO] [03/21/2016 22:41:51.119] [TypedActorSystem-akka.actor.default-dispatcher-2] [$Proxy0(akka://TypedActorSystem)] TypedActor  Started
[INFO] [03/21/2016 22:41:51.123] [TypedActorSystem-akka.actor.default-dispatcher-2] [$Proxy1(akka://TypedActorSystem)] TypedActor  Started
[INFO] [03/21/2016 22:41:51.124] [main] [Example12_10$(akka://TypedActorSystem)] fSquare=100
[INFO] [03/21/2016 22:41:51.131] [TypedActorSystem-akka.actor.default-dispatcher-5] [$Proxy1(akka://TypedActorSystem)] TypedActor Stopped
[INFO] [03/21/2016 22:41:51.131] [TypedActorSystem-akka.actor.default-dispatcher-3] [$Proxy0(akka://TypedActorSystem)] TypedActor Stopped

代码中类SquarerImpl 混入了PreStart和PostStop两个trait:class SquarerImpl(val name: String) extends Squarer with PostStop with PreStart,这样的话在创建TypedActor之前和停止TypedActor后能够进行相应的操作,本例中主要是为监视TypedActor的创建和停止过程。代码TypedActor(system).stop(mySquarer)通过stop方法停止TypedActor,而TypedActor(system)

.poisonPill(otherSquarer)通过调用poisonPill方法停止运行TypedActor。

时间: 2024-11-07 21:34:37

Akka并发编程——第八节:Actor模型(七)的相关文章

68:Scala并发编程原生线程Actor、Cass Class下的消息传递和偏函数实战解析及其在Spark中的应用源码解析

今天给大家带来的是王家林老师的scala编程讲座的第68讲:Scala并发编程原生线程Actor.Cass Class下的消息传递和偏函数实战解析 昨天讲了Actor的匿名Actor及消息传递,那么我们今天来看一下原生线程Actor及CassClass下的消息传递,让我们从代码出发: case class Person(name:String,age:Int)//定义cass Class class HelloActor extends Actor{//预定义一个Actor  def act()

Scala 深入浅出实战经典 第68讲:Scala并发编程原生线程Actor、Cass Class下的消息传递和偏函数实战解析

王家林亲授<DT大数据梦工厂>大数据实战视频 Scala 深入浅出实战经典(1-87讲)完整视频.PPT.代码下载: 百度云盘:http://pan.baidu.com/s/1c0noOt6 腾讯微云:http://url.cn/TnGbdC 360云盘:http://yunpan.cn/cQ4c2UALDjSKy 访问密码 45e2土豆:http://www.tudou.com/programs/view/mm3eDHk3T5o/优酷:http://v.youku.com/v_show/id

Akka并发编程——第七节:Actor模型(六)

主要内容: 1. Typed Actor定义 2. Typed Actor创建 3. 消息发送 1. Typed Actor定义 Akka中的Typed Actor是Active Objects设计模式的实现,Active Objects模式将方法的执行和方法的调用进行解耦合,从而为程序引入并发性.Typed Actor由公用的接口和对应实现两部分构成,其后面深层次的实现使用的是代理模式,即通过使用JDK中的动态代理来实现,在调用接口的方法时自动分发到实现接口的对象上.Typed Actor的定

java并发编程12.java内存模型

假设一个线程为变量赋值:variable = 3: 内存模型需要解决一个问题:“在什么条件下,读取variable的线程将看到这个值为3?” 这看上去理所当然,但是如果缺少同步,那么将会有许多因素使得线程无法立即甚至永远,看到另一个线程的操作结果. 如: 1.在编译器中生成的指令顺序,可以与源代码中的顺序不同,此外编译器还会将变量保存在寄存器而不是内存中: 2.处理器可以采用乱序或并行等方式来执行指令: 3.缓存可能会改变将写入变量提交到主内存的次序: 4.而且保存在处理器本地缓存中的值,对于其

5 并发编程--队列&amp;生产者消费者模型

1.队列的介绍 进程彼此之间互相隔离,要实现进程间通信(IPC),multiprocessing模块支持两种形式:队列和管道,这两种方式都是使用消息传递的 创建队列的类(底层就是以管道和锁定的方式实现): Queue([maxsize]):创建共享的进程队列,Queue是多进程安全的队列,可以使用Queue实现多进程之间的数据传递. 参数介绍: maxsize是队列中允许最大项数,省略则无大小限制. 但需要明确: 1.队列内存放的是消息而非大数据 2.队列占用的是内存空间,因而maxsize即便

3.java并发编程艺术-java内存模型

3.1 java内存模型的基础 3.1.1并发编程模型的两个关键问题 在并发编程中,需要处理两个关键问题:线程之间如何通信及线程之间如何同步(这里的线程是指并发执行的活动实体).通信是指两个线程 之间以何种机制来交换信息.在命令式编程中,线程之间的通信机制有两种:共享内存和消息传递. 在共享内存的并发模型里,线程之间共享程序的公共状态,通过写-读内存中的公共状态进行隐式通信.在消息传递的并发模型里,线程之间没有公共状态,线程之间必须通过发送消息来进行显式进行通信. 同步是指程序中用于控制不同线程

scala并发编程原生线程Actor、Case Class下的消息传递和偏函数实战

参考代码: import scala.actors._ case class Person(name:String,age:Int) class HelloActor extends Actor{ def act(){ while(true){ receive{ case Person(name,age)=>{ //偏函数 println("Name: "+ name + ":" +"Age:" +age) sender ! "E

并发编程 之 生产者消费者模型

1 什么是生产者消费者模型 生产者:比喻的是程序中负责产生数据的任务 消费者:比喻的是程序中负责处理数据的任务 生产者->共享的介质(队列)<-消费者 2 为何用 实现了生产者与消费者的解耦和,生产者可以不停地生产,消费者也可以不停地消费 从而平衡了生产者的生产能力与消费者消费能力,提升了程序整体运行的效率 什么时候用? 当我们的程序中存在明显的两类任务,一类负责产生数据,另外一类负责处理数据 此时就应该考虑使用生产者消费者模型来提升程序的效率 from multiprocessing imp

并发编程: 生产消费模型、死锁与Rlock、线程、守护线程、信号量、锁

一.守护进程 二.互斥锁 三.抢票 四.进程间通讯 五.进程间通讯2 一.守护进程 """ 进程间通讯的另一种方式 使用queue queue 队列 队列的特点: 先进的先出 后进后出 就像扶梯 """ from multiprocessing import Process,Queue # 基础操作 必须要掌握的 # 创建一个队列 # q = Queue() # # 存入数据 # q.put("hello") # q.put(