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

主要内容:
1. Typed Actor定义
2. Typed Actor创建
3. 消息发送

1. Typed Actor定义

Akka中的Typed Actor是Active Objects设计模式的实现,Active Objects模式将方法的执行和方法的调用进行解耦合,从而为程序引入并发性。Typed Actor由公用的接口和对应实现两部分构成,其后面深层次的实现使用的是代理模式,即通过使用JDK中的动态代理来实现,在调用接口的方法时自动分发到实现接口的对象上。Typed Actor的定义[ ]如下所示。

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
  }

  class SquarerImpl(val name: String) extends Squarer {
    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
  }

trait Squarer中定义了4个方法:
(1)def squareDontCare(i: Int): Unit方法:返回值类型为Unit,它类似于Untyped Actor中的fire-and-forget消息发送模型,即!和tell方法调用。
(2)def square(i: Int): Future[Int]:返回值类型为Future[Int],它类似于Untyped Actor中的send-request-reply消息发送模型,即?和ask方法调用,此种调用是非阻塞的。
(3)def squareNowPlease(i: Int): Option[Int]:返回值类型为Option[Int](Option类可以是scala.Option[_]也可以是akka.japi.Option

2. 创建Typed Actor

通过下列代码创建Typed Actor实例。

//直接通过默认的构造函数创建Typed Actor
val mySquarer: Squarer =TypedActor(system).typedActorOf(TypedProps[SquarerImpl]())
//直接通过默认的构造函数创建Typed Actor并指定Typed Actor名称
val mySquarer: Squarer =TypedActor(system).typedActorOf(TypedProps[SquarerImpl](),"mySquarer")
//通过非默认的构造函数创建Typed Actor并指定Typed Actor名称
val otherSquarer: Squarer = TypedActor(system).typedActorOf(TypedProps(classOf[Squarer],new SquarerImpl("SquarerImpl")), "otherSquarer")

上面代码演示的是使用构造函数和非默认构造函数创建Typed Actor,其中Squarer为代理的类型,SquarerImpl为具体实现的类型。

3. 消息发送

//fire-forget消息发送
  mySquarer.squareDontCare(10)

  //send-request-reply消息发送
  val oSquare = mySquarer.squareNowPlease(10)

  val iSquare = mySquarer.squareNow(10)

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

代码mySquarer.squareDontCare(10)是单向消息发送,方法将在另外一个线程上异步地执行;val oSquare = mySquarer.squareNowPlease(10)、val iSquare = mySquarer.squareNow(10)为Request-reply消息发送,在特定时间内以阻塞的方式执行,对于.squareNowPlease(10)方法如果在对应时间内没有返回结果则返回值为None,否则返回值为Option[Int]类型,对于squareNow(10)方法如果在对应时间内无返回值则会抛出异常java.util.concurrent.TimeoutException,否则返回Int类型值;val fSquare = mySquarer.square(10)为Request-reply-with-future式的消息发送,以非阻塞的方式执行,可以通过val result = Await.result(fSquare, 5 second)获取执行结果。完整代码如下所示。

/*
 * Typed Actor
 */
object Example_01 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
  }

  class SquarerImpl(val name: String) extends Squarer {
    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
  }

  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")

  //fire-forget消息发送
  mySquarer.squareDontCare(10)

  //send-request-reply消息发送
  val oSquare = mySquarer.squareNowPlease(10)

  log.info("oSquare="+oSquare)

  val iSquare = mySquarer.squareNow(10)
  log.info("iSquare="+iSquare)

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

  log.info("fSquare="+result)

  system.shutdown()
}

代码运行结果如下:
[INFO] [03/21/2016 21:15:50.592] [main] [Example12_9(akka://TypedActorSystem)]oSquare=Some(100)[INFO][03/21/201621:15:50.649][main][Example129(akka://TypedActorSystem)] iSquare=100
[INFO] [03/21/2016 21:15:50.649] [main] [Example12_9$(akka://TypedActorSystem)] fSquare=100

时间: 2025-01-21 12:32:25

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

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. 具体使用代码

Akka并发编程——第三节:Actor模型(二)

本节主要内容: Actor API解析 1. Actor API解析 Actor中的主要成员变量和方法定义如下: package akka.actor trait Actor extends scala.AnyRef { type Receive = akka.actor.Actor.Receive //context变量暴露当前Actor的上下文信息及当前消息 implicit val context : akka.actor.ActorContext = { /* compiled code

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

本将主要内容: 1. Actor引用.Actor路径 1. Actor引用.Actor路径 下图是Akka官方文档中给出的一张图 该图清晰地说明了ActorPath,ActorRef,Actor及ActorSystem之间的关系,并说明了Actor整体的层次结构.前面我们提到,Akka应用程序会持有一个名称为user的Actor,该Actor被称为guardian supervisor(守卫监督器),无论是ActorSystem创建的Actor还是通过ActorContext创建的Actor都为

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

本节主要内容: 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 w

Akka v2.4.11 发布,Actor 模型开发库

Akka v2.4.11 发布了. Akka 是一个用 Scala 编写的库,用于简化编写容错的.高可伸缩性的 Java 和 Scala 的 Actor 模型应用. 改进内容,请点击此处查看.更多详情请关注发行日志和提交记录. 文章转载自 开源中国社区 [http://www.oschina.net]

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

本将主要内容: 1. !消息发送,Fire-and-Forget消息模型 2. ?消息发送,Send-And-Receive-Future消息模型 Akka提供了两种消息模型:fire-and-forget和Send-And-Receive-Future.fire-and-forget是一种单向消息发送模型,指的是异步发送消息,通过异步发送消息且消息发送后可以立即返回,Akka中使用?方法进行fire-and-forget消息发送,如stringActor!"Creating Actors wi

Akka并发编程——第二节:Actor模型(一)

本节主要内容 定义Actor 创建Actor 1. 定义Actor 通过扩展akka.actor.Actor 特质并实现receive方法来定义Actor,代码示例如下 //通过扩展Actor并实现receive方法来定义Actor class MyActor extends Actor { //获取LoggingAdapter,用于日志输出 val log = Logging(context.system, this) //实现receive方法,定义Actor的行为逻辑,返回的是一个偏函数

Java并发编程示例(七):守护线程的创建和运行_java

Java有一种特殊线程,守护线程,这种线程优先级特别低,只有在同一程序中的其他线程不执行时才会执行. 由于守护线程拥有这些特性,所以,一般用为为程序中的普通线程(也称为用户线程)提供服务.它们一般会有一个无限循环,或用于等待请求服务,或用于执行任务等.它们不可以做任何重要的工作,因为我们不确定他们什么时才能分配到CPU运行时间,而且当没有其他线程执行时,它们就会自动终止.这类线程的一个典型应用就是Java的垃圾回收. 在本节示例中,我们将创建两个线程,一个是普通线程,向队列中写入事件:另外一个是

Akka框架——第一节:并发编程简介

本节主要内容: 1. 重要概念 2. Actor模型 3. Akka架构简介 多核处理器的出现使并发编程(Concurrent Programming)成为开发人员必备的一项技能,许多现代编程语言都致力于解决并发编程问题.并发编程虽然能够提高程序的性能,但传统并发编程的共享内存通信机制对开发人员的编程技能要求很高,需要开发人员通过自身的专业编程技能去避免死锁.互斥等待及竞争条件(Race Condition)等,熟悉Java语言并发编程的读者们对这些问题的理解会比较深刻,这些问题使得并发编程比顺