Simple Introduction On Akka Actor

fujohnwang

2011-04-29

author: fujohnwang

Simple Introduction就是简单的介绍， 呵呵，所以， 这篇文字只是初级的浮于表面的泛泛之谈， 不要期望写的有多么事无巨细， 写得有多么深入人心，简单了解一下吧，感兴趣自己再找资料啃呗！

Akka Framework有一堆modules， 这里只扯它的Actor支持， 什么不知道Actor是啥？哦， 那先去看看Erlanga啦， Wikipedia啦， Scala Actor啦， 然后再回来看这篇文字，呵呵，因为我不会那么详细的告诉你Actor到底是个什么东西， ：）

Actor Vs. Task

Actor和Task那， 可以说是Share Nothing/Message Passing Concurrency跟Shared State Concurrency的标志性 “建筑物” ， 要比较很细的话， 可能也是一家有一家的话要说， 我那只说一点我的观点， 各位看官要是看得不爽， 也别喷， 欢迎纠正，使得我也可以迷途知返嘛！

我的观点，简单来说， Actor可以维护多次逻辑执行的状态， 而Task则不维护这种逻辑多次执行的状态。 举例来说， 你把一个Callable或者Runnable提交给ExecutorService，并且当这个Task(Callable和Runnable算是Task的某些具体形式)执行完成后， 它的状态就此完结， 不会对下一个提交并执行的Task产生任何影响（当然， task与task之间通过某种全局的状态管理的情况除外）； 而Actor则不同， 一笔按照Actor的逻辑执行后， 大部分会对下一笔程序同样逻辑执行产生影响，因为它们的执行状态是通过Actor本身来维护的。 本质上，这种情况算是基于Task的并发建模与基于数据的并发建模的差异所造成的。

扯完这些，我们下面看本篇文字的主角-Akka Actor。

Akka Actor Quick Start

Akka Actor是Scala平台上的一种Actor实现， 其它的还有Scala本身类库提供的默认Actor实现，Lift框架的Actor实现，等等， 但Akka相对于其它Actor实现来说， “忽悠” 的名声有点儿大（ 因为基本上所有Actor实现上， 一些基本的组件都会有的，只不过，它忽悠的也还好， 性能确实比Scala默认提供的Actor实现要好），所以这也就是为啥要简单介绍一下它的原因啦～

AkkaActor分两种， 一种叫Untyped Actor， 这种Actor就是常见的跟Erlang的ActorAPI实现和使用方式类似的那种， 而另一种叫Typed Actor, 它是通过API的强类型约束来声明和实现Actor， 后面会详细介绍它，我们先看Untyped Actor…

UnTyped Actor

Untyped Actor的实现很简单，只要实现akka.actor.Actor这个trait就行：

class MyActor extends Actor{
     protected def receive = {
       case msg:String=> println("receives message:"+msg)
       case _=> println("unexpected message")
     }
}

一个Untyped Actor，只要实现receive方法就行了， Akka会有外围的调度框架来调度和执行它。知道了怎么定义一个Akka Actor， 那下面就是怎么创建和使用它了。

Akka Actor的创建和启动

AkkaActor并不是直接实例化你的Actor定义并使用它，而是通过ActorRef来引用和管理你的Actor实例， 代码说话：

val myActor:ActorRef = actorOf[MyActor]
myActor.start

akka.actor.Actor.actorOf工厂方法会根据你的Actor定义的类型来实例化和启动相应的Actor实例， 返回的Actor实例是ActorRef类型，而不是Actor类型，这个与Scala的Actor是不同的地方。（以上myActor定义的时候声明的类型是为了强调， 实际上因为有Scala的类型推导，此处的类型是可以省略的） akka.actor.Actor.actorOf还有许多变体（重载）， 比如， 如果我们的Actor定义有构造方法参数：

class MyActor(id:String) extends Actor{
     protected def receive = {
       case msg:String=> println("receives message:"+msg)
       case _=> println("unexpected message")
     }
}

那么， 可以使用以下形式来实例化和启动Actor:

val myActor:ActorRef = actorOf(new Actor("my actor's id"))
myActor.start

更多构造方式可以参阅Akka的Scaladoc， 此处有在线版本: http://akka.io/api/1.0/akka-actor/

Tip
Actor的实例化和启动也可以以chaining的方式一次搞定:
    val myActor:ActorRef = actorOf(new Actor("my actor's id")).start

Akka Actor的使用

Actor是以消息传递(Message Passing)的方式来实现并发的，所以， 使用Actor基本上就可以简化为Actor间的发送消息， 与大部分Actor一样， AkkaActor提供了以下几种消息发送方式:

  • ! - 异步消息发送. (!在这里读“bang”， 单一的!表示fire-and-forget形式的异步消息发送，该方法会即可返回而不用等消息处理完毕， 属于最常见的actor操作，也是效率更高的。 异步意味着系统的各个部分和资源可以得到更有效的利用.)
  • !! - 异步消息发送并且同步等待. (bangbang可以表述为fire-and-wait, 消息也是异步处理，但知道消息处理完成之前， !!方法不会返回。 显然这种方式效率不如!的形式， 最好是迫不得已的特定场景下才选用它。)
  • !!! - 异步发送消息并返回Future. (bangbangbang与我们executor.submit返回一个future类似， 它发送完消息之后即可返回一个future， 拿到这个future之后，用户自己来决定等待还是先做其它事情。可以描述为fire-and-reply-asynchronously.)

为了说明以上说明场景，假设我们有这么一个Actor，它收到数字类型的消息，则直接异步处理； 收到字符串形式的消息， 则将其变为大写，并返回， 该Actor定义如下：

class MyActor extends Actor{
  protected def receive = {
    case n:Int => println("receives number:"+n)
    case msg:String=> self.reply(msg.toUpperCase)
    case _=> println("unexpected message")
  }
}

现在，要Actor开工:

val myActor:ActorRef = actorOf[MyActor].start
myActor ! 1                            //receives number:1
println((myActor !! "darren").getOrElse("impossible"))        // DARREN
val future = myActor !!! "darren"
// do something else
future.await
future.result.asInstanceOf[Option[String]] match{
   case Some(str)=> println("converted string:"+str)  // converted string:DARREN
   case None => println("impossible")
}

See？ Easy And Fun! 就这么简单， 别在问我其它的了，我也不知道 ：-）

Stop or KillTheActor

Actor干完活， 不能让它在哪儿浪费资源不是？！ 要关闭/停止Actor， 可以采用多种形式， 最常见的就是直接掉actor.stop, 其它几种方式还有:

1- Actor接收到特定消息后主动退出.

case object ShutdownSignal

class MyActor extends Actor{
  protected def receive = {
 case ShutdownSignal => exit
 case _=> ...
  }
}

2- 喂毒药. 每次定义个ShutdownSignal啥的， 还得每次都在Actor实现的时候处理， 真麻烦， 时不？ 其实，可以不这么干， 直接喂你的Actor， 毒死它，哈哈…. 只要myActor ! akka.actor.PoisonPill， 你懂的…

3- 通过ActorRegistry统一关闭. Actor.registry.shutdownAll， 你也懂的

Lifecycle Callbacks

除了启动和关闭Actor， Akka Actor还有其它几个有关其生命周期的回掉方法:

  • preStart
  • postStop
  • preRestart
  • postRestart

具体在啥时候触发，方法名已经很明了了， 至于用来干什么，也不用多说吧？！做做资源初始化啦， 资源清理啦，这是最常见的用途咯

Other Features

Akka Actor还有些比较有趣的特性，比如运行期间的Actor逻辑互换， 即在运行期间，将当前的处理逻辑替换为新的处理逻辑， 这可以通过Hotswap消息或者become方法来实现， 以Hotswap消息为例(摘自Akka Actor文档):

actor ! HotSwap( self => {
  case message => self.reply("hotswapped body")
})

当actor收到这个消息之后，就会使用新的处理逻辑来处理后继的消息。这个新的block会被放到最stack的最上面， 本着“自己的屁股自己擦”的原则，是不是用完后要处理掉那？ become和unbecome可能更能说明这个问题， 有创建就有销毁嘛！

class Swapper extends Actor {
 def receive = {
   case Swap =>
     println("Hi")
     become {
       case Swap =>
         println("Ho")
         unbecome // resets the latest 'become' (just for fun)
     }
 }
}

Hi， Ho， Hi，Ho…

Typed Actor

Typed Actor就是通过强类型的接口约束以及其POJO形式的实现而表现出来的一种Actor形式。Akka框架通过AOP对这些POJO实现相应的接口方法进行拦截和代理， 从而实现跟UntypeActor类似的异步消息处理。 反正方法调用本来也是消息传递的一种嘛。

Typed Actor是根据接口的方法定义的Signature来决定处理语义的， 比如是异步的消息处理，还是异步接收同步返回， 以以下接口为例：

interface RegistrationService {
  void register(User user, Credentials cred)
  User getUserFor(String username)
}

返回值为void的方法， 相当于Untyped Actor的！方法， 而有返回值的， 则相当于！！， 至于跟!!!相当的方法定义，就得返回特定的akka.dispatch.Future类型咯。

Typed Actor个人没有啥喜好，所以不做赘述， 感兴趣可以参考 官方文档 。

Dispatcher

所有的AkkaActor创建之后，如果没有特殊设置，都会使用DispatchersglobalExecutorBasedEventDrivenDispatcher进行运行调度， 从运行日志就可以看出: 
14:27:19.659 [main] DEBUG a.d.Dispatchers$globalExecutorBasedEventDrivenDispatcher$ 
   - Starting up Dispatchers$globalExecutorBasedEventDrivenDispatcher$[akka:event-driven:dispatcher:global]
 with throughput [5]

但很多时候， 所有的Actor使用同一个Dispatcher是远远不够的。

要定制相应Actor的运行调度， 可以为其设置特定的Dispatcher实现， 常见的Dispatcher实现在akka.dispatch包下， 如:

  • ExecutorBasedEventDrivenDispatcher
  • ExecutorBasedEventDrivenWorkStealingDispatcher
  • HawtDispatcher
  • 可以通过两种方式为我们的Actor指定自己的Dispatcher， 第一种， 直接在Actor定义中指定：
class MyActor extends Actor{
 self.dispatcher = ...
}

使用哪种现有的Dispatcher实现或者自己实现， 可以根据自己的当前环境来决定。 另一种， 可以在Actor的ActorRef上设置:

val myActor = actorOf(..)
myActor.dispatcher = ...
myActor.start

只要在Actor启动前设置即可， 个人倾向于这种方式。

Fault Tolerance Facility

为了保证系统的健壮性， Akka提供了类似于Erlang的Actor Supervisor的策略， 那个图我就不贴了，就是按照层次， 上层监控下层，如果下层有Actor挂掉了， 那么由上层的Supervisor来决定是重启这个Actor那，还是重启Supervisor当前监控的所有Actor， 所谓的OneForOne和AllForOne策略。

基于Supervisor的策略的基本原则是Let it crash, 当目标Actor crash掉之后， 要么重启它或者相关的actors， 要么正常关闭它。所以， 有几点主要要素要考虑在内， 其一， 需要监控哪些条件来判断目标的Actor有没有crash， 这通常是通过Exception来确定的；其二， 当监控到目标Actor crash之后， 是要重启它，还是直接关闭它？ 其三， 当要重启它的时候，是只重其它那，还是重启跟它相关的所有Actors？ 本着这几个要素，Akka 提供了相应的设施来界定这些关注点。

要实现基于Supervisor的Fault Tolerance管理， Akka中通常采用所谓的声明式的实现方式， 以下是常见的代码：

val supervisor = Supervisor(
  SupervisorConfig(
    AllForOneStrategy(List(classOf[Exception]), 3, 1000),
    Supervise(
      actorOf[MyActor1],
      Permanent) ::
    Supervise(
      actorOf[MyActor2],
      Permanent) ::
    Nil))

Supervisor有一个company object提供了工厂方法来创建和启动一个Supervisor实例，即以上代码种的supervisor， 该工厂方法接受一个SupervisorConfig参数， 而SupervisorConfig参数可以定义Fault Handling的策略，以及要对哪些actor进行监控， 在以上代码中， AllForOneStrategy(List(classOf[Exception]), 3, 1000)定义了当捕获到目标Actor抛出Exception之后， 重启当前supervisor监控的所有actors， 并且最多重启3次， 时间不超出1秒钟。该项配置阐述了我们之前提到的第一和第三要素。SupervisorConfig第二个参数是接受了一组Supervise的List， 每个Supervise定义一个被当前supervisor监控的actor信息， 比如要监控的actor实例引用，以及该actor crash掉之后， 只要一直重启还是只是正常关闭它， 这阐述了我们之前提到的第二个要素。

通过Supervisor创建Supervisor， 最终创建后的实例是已经start的， 不需要显式的start； 不过， 如果需要控制实例创建的时机以及start的时机， 可以使用akka.actor.SupervisorFactory类:

val factory = SupervisorFactory(
   SupervisorConfig(
     AllForOneStrategy(OneForOne, 3, 10, List(classOf[Exception]),
     Supervise(
       myFirstActor,
       Permanent) ::
     Supervise(
       mySecondActor,
       Permanent) ::
     Nil))  
val supervisor = factory.newInstance
supervisor.start // start up all managed server

不过，大多数情况下，使用Supervisor就OK啦。

以下是一个完整的实例， 可以一窥声明式的Fault Handling是多么的简单和容易理解：

class MyActor extends Actor {
 // self.setLifeCycle(akka.config.Supervision.permanent)
      override def postRestart(reason: Throwable) {
        println("restart is done.")
      }
      protected def receive = {
        case n: Int => println("receives number:" + n)
        case msg: String => self.reply(msg.toUpperCase)
        case _ => throw new Exception("xxxx")
      }
    }
    import akka.config.Supervision._
    val myActor: ActorRef = actorOf[MyActor].start
    val supervisor = Supervisor(SupervisorConfig(
      AllForOneStrategy(List(classOf[Exception]), 3, 1000),
      Supervise(
        myActor,
        Permanent) ::
        Nil))
 // supervisor.link(myActor)
    myActor ! Array[Byte](0x00)
    myActor ! 111

    TimeUnit.SECONDS.sleep(5)

    supervisor.shutdown
    myActor.stop

在myActor ! ArrayByte之后， myActor会抛出异常， 但因为有supervisor监控它并重启了它，所以myActor ! 111可以正常处理。 如果不使用声明式的fault handling， 那么需要在supervisor以及被监控actor内部各自声明一些钩子， 以上代码实例中被注视的代码算是一部分， 使用了声明式的方式，这些就都不需要了。

要了解fault handling更多信息，包括编程式的fault handling方式， 可以参考 这里 .

Remote Actor

使用就不扯了，简单说一下原理，不过我自己瞎猜的，没看它代码，呵呵， RemoteActor要实现起来其实很简单， 直接对ActorRef做Proxy， 发送端结点与接收端结点采用不同的逻辑实现， 即接收端结点起相应的TCP服务接收请求， 然后对接收到的请求进行分类并分发给相应的本地Actor， 如果是异步消息，则返回；否则等处理完后， 再将处理结果通过TCP连接发送回发送端结点；发送端结点的Proxy直接发起TCP连接到目标接收端， 之后，只要本地Actor收到请求，就通过该TCP连接将消息发送给目标端TCP Server，server处理逻辑前面已经说了。

在这里， 如果设计好的话， tcp的交互协议可以扩展， tcp的通信序列化协议也可以扩展， 比如， 针对Akka不支持protostuff的现状， 插入自定义的protostuff序列化装置就很好嘛，哈哈

Conclusion

AkkaActor相对于其它的实现来说更成熟一些， 但没有最好，只有更好，希望Scala平台的Actor更加成熟，更加快速，更加强大…

>>>>>> 更多阅读 <<<<<<

「福强私学」来一个？

「福强私学」, 一部沉淀了个人成长、技术与架构、组织与管理以及商业上的方法与心法的百科全书。

footer img for kb.afoo.me

开天窗，拉认知，订阅「福报」，即刻拥有自己的全模态人工智能。

订阅「福报」