Akka源码分析-Actor&ActorContext&ActorRef&ActorCell

Posted 2020-12-25 gabry

　　分析源码的过程中我们发现，Akka出现了Actor、ActorRef、ActorCell、ActorContext等几个相似的概念，它们之间究竟有什么区别和联系呢？

/**
 * Actor base trait that should be extended by or mixed to create an Actor with the semantics of the ‘Actor Model‘:
 * <a href="http://en.wikipedia.org/wiki/Actor_model">http://en.wikipedia.org/wiki/Actor_model</a>
 *
 * An actor has a well-defined (non-cyclic) life-cycle.
 *  - ‘‘RUNNING‘‘ (created and started actor) - can receive messages
 *  - ‘‘SHUTDOWN‘‘ (when ‘stop‘ is invoked) - can‘t do anything
 *
 * The Actor‘s own [[akka.actor.ActorRef]] is available as `self`, the current
 * message’s sender as `sender()` and the [[akka.actor.ActorContext]] as
 * `context`. The only abstract method is `receive` which shall return the
 * initial behavior of the actor as a partial function (behavior can be changed
 * using `context.become` and `context.unbecome`).
 *
 * This is the Scala API (hence the Scala code below), for the Java API see [[akka.actor.AbstractActor]].
 *
 * {{{
 * class ExampleActor extends Actor {
 *
 *   override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
 *     case _: ArithmeticException      => Resume
 *     case _: NullPointerException     => Restart
 *     case _: IllegalArgumentException => Stop
 *     case _: Exception                => Escalate
 *   }
 *
 *   def receive = {
 *                                      // directly calculated reply
 *     case Request(r)               => sender() ! calculate(r)
 *
 *                                      // just to demonstrate how to stop yourself
 *     case Shutdown                 => context.stop(self)
 *
 *                                      // error kernel with child replying directly to ‘sender()‘
 *     case Dangerous(r)             => context.actorOf(Props[ReplyToOriginWorker]).tell(PerformWork(r), sender())
 *
 *                                      // error kernel with reply going through us
 *     case OtherJob(r)              => context.actorOf(Props[ReplyToMeWorker]) ! JobRequest(r, sender())
 *     case JobReply(result, orig_s) => orig_s ! result
 *   }
 * }
 * }}}
 *
 * The last line demonstrates the essence of the error kernel design: spawn
 * one-off actors which terminate after doing their job, pass on `sender()` to
 * allow direct reply if that is what makes sense, or round-trip the sender
 * as shown with the fictitious JobRequest/JobReply message pair.
 *
 * If you don’t like writing `context` you can always `import context._` to get
 * direct access to `actorOf`, `stop` etc. This is not default in order to keep
 * the name-space clean.
 */
trait Actor {

  // to make type Receive known in subclasses without import
  type Receive = Actor.Receive

  /**
   * Scala API: Stores the context for this actor, including self, and sender.
   * It is implicit to support operations such as `forward`.
   *
   * WARNING: Only valid within the Actor itself, so do not close over it and
   * publish it to other threads!
   *
   * [[akka.actor.ActorContext]] is the Scala API. `getContext` returns a
   * [[akka.actor.AbstractActor.ActorContext]], which is the Java API of the actor
   * context.
   */
  implicit val context: ActorContext = {
    val contextStack = ActorCell.contextStack.get
    if ((contextStack.isEmpty) || (contextStack.head eq null))
      throw ActorInitializationException(
        s"You cannot create an instance of [${getClass.getName}] explicitly using the constructor (new). " +
          "You have to use one of the ‘actorOf‘ factory methods to create a new actor. See the documentation.")
    val c = contextStack.head
    ActorCell.contextStack.set(null :: contextStack)
    c
  }

  /**
   * The ‘self‘ field holds the ActorRef for this actor.
   * <p/>
   * Can be used to send messages to itself:
   * <pre>
   * self ! message
   * </pre>
   */
  implicit final val self = context.self //MUST BE A VAL, TRUST ME

  /**
   * The reference sender Actor of the last received message.
   * Is defined if the message was sent from another Actor,
   * else `deadLetters` in [[akka.actor.ActorSystem]].
   *
   * WARNING: Only valid within the Actor itself, so do not close over it and
   * publish it to other threads!
   */
  final def sender(): ActorRef = context.sender()

  /**
   * Scala API: This defines the initial actor behavior, it must return a partial function
   * with the actor logic.
   */
  //#receive
  def receive: Actor.Receive
  //#receive

  /**
   * INTERNAL API.
   *
   * Can be overridden to intercept calls to this actor‘s current behavior.
   *
   * @param receive current behavior.
   * @param msg current message.
   */
  @InternalApi
  protected[akka] def aroundReceive(receive: Actor.Receive, msg: Any): Unit = {
    // optimization: avoid allocation of lambda
    if (receive.applyOrElse(msg, Actor.notHandledFun).asInstanceOf[AnyRef] eq Actor.NotHandled) {
      unhandled(msg)
    }
  }

  /**
   * INTERNAL API.
   *
   * Can be overridden to intercept calls to `preStart`. Calls `preStart` by default.
   */
  @InternalApi
  protected[akka] def aroundPreStart(): Unit = preStart()

  /**
   * INTERNAL API.
   *
   * Can be overridden to intercept calls to `postStop`. Calls `postStop` by default.
   */
  @InternalApi
  protected[akka] def aroundPostStop(): Unit = postStop()

  /**
   * INTERNAL API.
   *
   * Can be overridden to intercept calls to `preRestart`. Calls `preRestart` by default.
   */
  @InternalApi
  protected[akka] def aroundPreRestart(reason: Throwable, message: Option[Any]): Unit = preRestart(reason, message)

  /**
   * INTERNAL API.
   *
   * Can be overridden to intercept calls to `postRestart`. Calls `postRestart` by default.
   */
  @InternalApi
  protected[akka] def aroundPostRestart(reason: Throwable): Unit = postRestart(reason)

  /**
   * User overridable definition the strategy to use for supervising
   * child actors.
   */
  def supervisorStrategy: SupervisorStrategy = SupervisorStrategy.defaultStrategy

  /**
   * User overridable callback.
   * <p/>
   * Is called when an Actor is started.
   * Actors are automatically started asynchronously when created.
   * Empty default implementation.
   */
  @throws(classOf[Exception]) // when changing this you MUST also change ActorDocTest
  //#lifecycle-hooks
  def preStart(): Unit = ()

  //#lifecycle-hooks

  /**
   * User overridable callback.
   * <p/>
   * Is called asynchronously after ‘actor.stop()‘ is invoked.
   * Empty default implementation.
   */
  @throws(classOf[Exception]) // when changing this you MUST also change ActorDocTest
  //#lifecycle-hooks
  def postStop(): Unit = ()

  //#lifecycle-hooks

  /**
   * Scala API: User overridable callback: ‘‘‘By default it disposes of all children and then calls `postStop()`.‘‘‘
   * @param reason the Throwable that caused the restart to happen
   * @param message optionally the current message the actor processed when failing, if applicable
   * <p/>
   * Is called on a crashed Actor right BEFORE it is restarted to allow clean
   * up of resources before Actor is terminated.
   */
  @throws(classOf[Exception]) // when changing this you MUST also change ActorDocTest
  //#lifecycle-hooks
  def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    context.children foreach { child ?
      context.unwatch(child)
      context.stop(child)
    }
    postStop()
  }

  //#lifecycle-hooks

  /**
   * User overridable callback: By default it calls `preStart()`.
   * @param reason the Throwable that caused the restart to happen
   * <p/>
   * Is called right AFTER restart on the newly created Actor to allow reinitialization after an Actor crash.
   */
  @throws(classOf[Exception]) // when changing this you MUST also change ActorDocTest
  //#lifecycle-hooks
  def postRestart(reason: Throwable): Unit = {
    preStart()
  }
  //#lifecycle-hooks

  /**
   * User overridable callback.
   * <p/>
   * Is called when a message isn‘t handled by the current behavior of the actor
   * by default it fails with either a [[akka.actor.DeathPactException]] (in
   * case of an unhandled [[akka.actor.Terminated]] message) or publishes an [[akka.actor.UnhandledMessage]]
   * to the actor‘s system‘s [[akka.event.EventStream]]
   */
  def unhandled(message: Any): Unit = {
    message match {
      case Terminated(dead) ? throw DeathPactException(dead)
      case _                ? context.system.eventStream.publish(UnhandledMessage(message, sender(), self))
    }
  }
}

 　　Actor这个特质是直接面向开发者的，这里我就直接贴出了官方源码。从代码来看，trait Actor提供了对消息的处理，actor生命周期接口的暴露，还有当前执行上下文的引用（配置、父子actor关系等）。简单来说，它就是面向开发者的，拥有开发者定义、使用actor的所有接口、字段和配置信息。

　　我们把trait Actor的字段和方法进行分类，有4类：

　　1.执行上下文信息。context:ActorContext、self:ActorRef、sender():ActorRef

　　2.生命周期管理。aroundReceive、aroundPreStart、aroundPostStop、aroundPreRestart、aroundPostRestart、preStart、postStop、preRestart、postRestart

　　3.actor行为定义。receive、unhandled

　　4.监督策略。supervisorStrategy: SupervisorStrategy

 　　在trait Actor中有一个context: ActorContext，这是一个重要的字段，有必要对其进行分析。

implicit val context: ActorContext = {
    val contextStack = ActorCell.contextStack.get
    if ((contextStack.isEmpty) || (contextStack.head eq null))
      throw ActorInitializationException(
        s"You cannot create an instance of [${getClass.getName}] explicitly using the constructor (new). " +
          "You have to use one of the ‘actorOf‘ factory methods to create a new actor. See the documentation.")
    val c = contextStack.head
    ActorCell.contextStack.set(null :: contextStack)
    c
  }

　　下面是contextStack的定义。

val contextStack = new ThreadLocal[List[ActorContext]] {
    override def initialValue: List[ActorContext] = Nil
  }

 　　综合上下文，简单来说context就是取出当前ActorCell.contextStack的head并返回，然后设置head为null，但context的实现方式值得研究。Actor是一个trait，里面定义了字段，该字段只有在初始化的时候才知道确定的值。之前我们分析过，Actor的实例化也是通过接收消息（Create）异步完成的。那么为啥是通过ThreadLocal来赋值呢？按我的理解，因为这是在另一个线程中完成初始化并赋值的，线程间共享变量就用了ThreadLocal；另外还为了防止用户直接通过new来创建Actor，因为如果直接new的话，当前线程应该是没有给ActorCell.contextStack赋值的，也无法赋值，它是一个私有变量，所以就只能通过接收到创建Actor并给context赋值。那么究竟是在哪里赋值的呢？

//This method is in charge of setting up the contextStack and create a new instance of the Actor
  protected def newActor(): Actor = {
    contextStack.set(this :: contextStack.get)
    try {
      behaviorStack = emptyBehaviorStack
      val instance = props.newActor()

      if (instance eq null)
        throw ActorInitializationException(self, "Actor instance passed to actorOf can‘t be ‘null‘")

      // If no becomes were issued, the actors behavior is its receive method
      behaviorStack = if (behaviorStack.isEmpty) instance.receive :: behaviorStack else behaviorStack
      instance
    } finally {
      val stackAfter = contextStack.get
      if (stackAfter.nonEmpty)
        contextStack.set(if (stackAfter.head eq null) stackAfter.tail.tail else stackAfter.tail) // pop null marker plus our context
    }
  }

 　　有一个非常奇怪的地方，contextStack是一个List，按我的理解这个既然是ThreadLocal了，每个线程get/set都是相对独立的，互不影响，直接set当前值应该也是可以的。查了一下这段代码的提交历史，好像一直都是一个List或stack，且与Akka的官方维护人员沟通，也没有明确的答案。

　　通过上面的代码我们知道是把当前的this赋值给了context，而this是一个ActorCell，ActorCell继承了AbstractActor.ActorContext，AbstractActor.ActorContext继承了akka.actor.ActorContext，而trait Actor中的context就是一个akka.actor.ActorContext，所以赋值没有问题。

　　从ActorContext的定义以及赋值逻辑来看，它就是ActorCell的一个视图，或者说一个切面。也就是说开发者在继承Actor的时候，能够通过ActorContext获取ActorCell的当前信息（例如字段值）或能力（例如become当前actor行为）。另外ActorContext还继承了ActorRefFactory，还具有创建或停止actor的能力。

　　Actor和ActorContext我们基本就分析完毕了，ActorCell之前也分析的差不多了，下面我们来看看ActorRef。

abstract class ActorRef extends java.lang.Comparable[ActorRef] with Serializable {
  scalaRef: InternalActorRef ?

  /**
   * Returns the path for this actor (from this actor up to the root actor).
   */
  def path: ActorPath

  /**
   * Comparison takes path and the unique id of the actor cell into account.
   */
  final def compareTo(other: ActorRef) = {
    val x = this.path compareTo other.path
    if (x == 0) if (this.path.uid < other.path.uid) -1 else if (this.path.uid == other.path.uid) 0 else 1
    else x
  }

  /**
   * Sends the specified message to this ActorRef, i.e. fire-and-forget
   * semantics, including the sender reference if possible.
   *
   * Pass [[akka.actor.ActorRef]] `noSender` or `null` as sender if there is nobody to reply to
   */
  final def tell(msg: Any, sender: ActorRef): Unit = this.!(msg)(sender)

  /**
   * Forwards the message and passes the original sender actor as the sender.
   *
   * Works, no matter whether originally sent with tell/‘!‘ or ask/‘?‘.
   */
  def forward(message: Any)(implicit context: ActorContext) = tell(message, context.sender())

  /**
   * INTERNAL API
   * Is the actor shut down?
   * The contract is that if this method returns true, then it will never be false again.
   * But you cannot rely on that it is alive if it returns false, since this by nature is a racy method.
   */
  @deprecated("Use context.watch(actor) and receive Terminated(actor)", "2.2")
  private[akka] def isTerminated: Boolean

  final override def hashCode: Int = {
    if (path.uid == ActorCell.undefinedUid) path.hashCode
    else path.uid
  }

  /**
   * Equals takes path and the unique id of the actor cell into account.
   */
  final override def equals(that: Any): Boolean = that match {
    case other: ActorRef ? path.uid == other.path.uid && path == other.path
    case _               ? false
  }

  override def toString: String =
    if (path.uid == ActorCell.undefinedUid) s"Actor[${path}]"
    else s"Actor[${path}#${path.uid}]"
}

 　　首先ActorRef是可以序列化的，这是一个很重要的特性，意味着可以跨JVM、跨网络传输，这有什么用呢？如果你对Akka有点了解的话，一定会知道位置透明是它的特性之一。ActorRef是实现位置透明的重要的技术点。

　　细心的读者会发现，ActorRef字段和函数并不多。只有一个字段path:ActorPath，代表actor的路径，ActorPath相信你一定也有了解，这是标志当前actor在actor树形层级的位置，当然还有其他信息（例如所在节点、协议、和当前actor的uid）。仅有的几个函数比较重要的也就是tell/forward/isTerminated，其中isTerminated还被废弃了。那也就意味着，Actor的功能大概仅限于：发送消息、序列化传输、获取当前ActorPath。

　　不过还要一点需要说明：scalaRef: InternalActorRef ?。这个类型限定意味着混入ActorRef的特质必须扩展自InternalActorRef。其实Scala的这个特性我是不太喜欢的，既然要求混入ActorRef的子类必须扩展自InternalActorRef，那还不如直接把ActorRef的接口放到InternalActorRef或者反过来。不过估计还是为了封装吧。

/**
 * Internal trait for assembling all the functionality needed internally on
 * ActorRefs. NOTE THAT THIS IS NOT A STABLE EXTERNAL INTERFACE!
 *
 * DO NOT USE THIS UNLESS INTERNALLY WITHIN AKKA!
 */
private[akka] abstract class InternalActorRef extends ActorRef with ScalaActorRef { this: ActorRefScope ?

　　之前我们分析过，ActorSystem最终通过LocalActorRefProvider的actorOf函数返回的应该就是InternalActorRef的子类RepointableActorRef。通过源码我们知道，InternalActorRef又提供了以下几类信息或功能：Actor生命周期管理（start/resume/suspend/restart/stop/sendSystemMessage）、父子actor信息、provider信息。当然还有是否为本地actor的标志。本地actor往往意味着它的邮箱可以直接获取到。

　　我们还需要继续对InternalActorRef的子类ActorRefWithCell进行分析，因为它有可能是联系ActorCell的关键。

/**
 * Common trait of all actor refs which actually have a Cell, most notably
 * LocalActorRef and RepointableActorRef. The former specializes the return
 * type of `underlying` so that follow-up calls can use invokevirtual instead
 * of invokeinterface.
 */
private[akka] abstract class ActorRefWithCell extends InternalActorRef { this: ActorRefScope ?
  def underlying: Cell
  def children: immutable.Iterable[ActorRef]
  def getSingleChild(name: String): InternalActorRef
}

 　　根据官方注释及其命名，我们知道ActorRefWithCell是一个继承了InternalActorRef，同时又拥有Cell功能的抽象类（underlying字段）。RepointableActorRef就是直接继承ActorRefWithCell的。前文我们也分析过RepointableActorRef在初始化的时候对underlying赋值的，也就意味着RepointableActorRef拥有ActorCell的实例。

　　上图我绘制了这几个概念之间的关系，通过分析我们知道，trait Actor拥有ActorContext实例；ActorContext其实就是ActorCell的一个视图或者说部分功能集合；ActorCell中同时拥有Actor、ActorRef的实例；ActorRef也拥有ActorCell的实例。

　　我们来简单概括一下这几个概念的功能区别，trait Actor是提供给开发者定义actor的行为的，开发者只需要实现具体的接口就可以实现actor生命周期和行为的定制化开发；ActorContext则提供了ActorCell的一个视图或功能集合，主要是为了隔离开发者对ActorCell的直接引用，毕竟ActorCell的内容太多，而且如果不加限制的使用，则会影响线程安全性；ActorCell则提供了一个actor全部的功能，初始化、生命周期管理、dispatcher、mailbox等，这些都是Akka框架开发者需要关注的内部实现逻辑；ActorRef则又提供了Actor对外开放的功能，发送消息等，主要是为了给其他开发者或Actor提供发送消息的能力。

　　怎么样，分析到这里是不是感觉Akka这几个概念还是比较清晰和严谨的呢。当然这也设计相对来说就复杂了点，不过也没办法，设计精良的框架，哪个不复杂呢？