Akka并发编程——第七节：Actor模型

Posted 2020-07-31 zhouzhihubeyond

主要内容：
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/2016 21:15:50.649] [main] [Example12_9$(akka://TypedActorSystem)] iSquare=100
[INFO] [03/21/2016 21:15:50.649] [main] [Example12_9$(akka://TypedActorSystem)] fSquare=100