Akka FSM 源代码分析

萧猛

<simonxiao@qq.com>

术语约定

状态名：

业务状态名称，是我们用来表达业务逻辑的状态名称

状态实例：

程序中实例化的一个状态对象，能够从中取得状态名

FSM Trait：

指 akka.actor.FSM 特质。这个特质也是你的程序中使用FSM应该混入的特质。

FSM Object：

FSM Trait 的伴生对象。这里定义了用于FSM Trait 的一些数据结构

事件处理函数：

某个状态下，收到事件之后的处理动作，其输入參数是事件，输出下一个状态实例。

转换处理函数：

从当前状态名A转换到下一个状态名B时须要运行的操作，其输入參数（A,B)的元组。无返回值。

状态机的装配机制

首先我们看一下 FSM Trait 的定义：

     trait FSM[S, D] extends Actor with Listeners with ActorLogging {// S 是状态名，我们定义的每个状态名都应该是这个S类型或者其子类型import FSM._type State = FSM.State[S, D] //状态实例的类型定义type StateFunction = scala.PartialFunction[Event, State] //事件处理函数类型定义type Timeout = Option[FiniteDuration]type TransitionHandler = PartialFunction[(S, S), Unit]   //转换处理函数类型定义   

我们先弄清楚前面说到的两个术语，状态名 和 状态实例。

FSM[S, D] 中的模板參数 S 是状态名的类型。我们定义的每个状态名都应该是这个S类型或者其子类型。 FSM.State[S, D]是定义在 FSM Object中的数据类型(一个case class)，它是真正在状态机实现中代表一个状态的实例。

251行对给这个类型定义了一个简称。

252行和254行定义了事件处理函数和转换处理函数的类型。注意：这两个函数的类型是 PartialFunction ，这意味着我们能用orElse方法把多个同类型函数串接起来。

为了解决这个问题Q1.1，我们须要设计数据结构来保存状态名、状态名相应的事件处理函数，还有转换处理函数。看代码：

     private val stateFunctions = mutable.Map[S, StateFunction]()private val stateTimeouts = mutable.Map[S, Timeout]()

     private var transitionEvent: List[TransitionHandler] = Nil

     private var currentState: State = _

这是几个私有的数据定义。stateFunctions 是以状态名为key 。事件处理函数为值创建的可改变的Map，初始值为空Map。这一个数据结构就保存了全部的状态名和事件处理函数。

每一个状态名相应一个可选的超时时间，保存在 stateTimeouts 映射中。

transitionEvent 保存了全部的转换处理函数。

currentState 是仅仅状态机当前状态的状态实例。初始值为 null 。

这几个重要的作用结构保存了状态机的静态结构，问题Q1.1解决。

Akka FSM 提供了一些内部DSL语法来协助装配状态机，也就是用来把你的状态机结构用前面的数据结构定义出来。

使用DSL机制有个特点。你要是明确DSL详细是如何干活的。使用起来就简单方便，心里非常踏实。要是不明确就会感觉有点神奇主义，用了也会不踏实。感觉没着没落的。

所以我们要把它弄明确。事实上也非常easy。

      final def when(stateName: S, stateTimeout: FiniteDuration = null)(stateFunction: StateFunction): Unit =register(stateName, stateFunction, Option(stateTimeout))

   private def register(name: S, function: StateFunction, timeout: Timeout): Unit = {if (stateFunctions contains name) {stateFunctions(name) = stateFunctions(name) orElse functionstateTimeouts(name) = timeout orElse stateTimeouts(name)} else {stateFunctions(name) = functionstateTimeouts(name) = timeout}}

    class Buncher extends Actor with FSM[State, Data] {startWith(Idle, Uninitialized)when(Idle) {case Event(SetTarget(ref), Uninitialized) =>stay using Todo(ref, Vector.empty)}// transition elided ...when(Active, stateTimeout = 1 second) {case Event(Flush | StateTimeout, t: Todo) =>goto(Idle) using t.copy(queue = Vector.empty)}// unhandled elided ...initialize()}

    final def startWith(stateName: S, stateData: D, timeout: Timeout = None): Unit =currentState = FSM.State(stateName, stateData, timeout)

   final def onTransition(transitionHandler: TransitionHandler): Unit = transitionEvent :+= transitionHandler

状态机的运转机制

正常状态事件的处理。这里要解决两个问题。Q2.1和Q2.2。也就是事件处理函数怎样被调用，状态转换函数怎样被调用。

FSM Trate 的 receive 方法的最后一个case语句处理正常的事件消息(上面的case处理定时器，listen等)。

    case value => {if (timeoutFuture.isDefined) {timeoutFuture.get.cancel()timeoutFuture = None}generation += 1processMsg(value, sender)}

value 是收到的Akka 原始消息。

    private def processMsg(value: Any, source: AnyRef): Unit = {val event = Event(value, currentState.stateData)processEvent(event, source)}

processMsg 将Akka原始消息封装成 状态事件 Event 调用 processEvent .

    private[akka] def processEvent(event: Event, source: AnyRef): Unit = {val stateFunc = stateFunctions(currentState.stateName)val nextState = if (stateFunc isDefinedAt event) {stateFunc(event)} else {// handleEventDefault ensures that this is always definedhandleEvent(event)}applyState(nextState)}

processEvent 函数很关键，先找到当前状态的事件处理函数（593行）。假设找到的事件处理函数可以处理收到的事件(594行)。那就调用事件处理函数（595行），不能处理该事件则做缺省处理(598行。后文再细说)。

事件处理函数会返回下一个状态实例 nextStat，然后调用 applyState 函数试图转换到下一个状态(600行)。

    private[akka] def applyState(nextState: State): Unit = {nextState.stopReason match {case None => makeTransition(nextState)case _ =>nextState.replies.reverse foreach { r => sender ! r }terminate(nextState)context.stop(self)}}

applyState 先检查目标状态是否是终止状态(604行)，假设不是则调用makeTransition 函数运行状态转换(605行)。否则启动终止流程(608行)。

    private[akka] def makeTransition(nextState: State): Unit = {if (!stateFunctions.contains(nextState.stateName)) {terminate(stay withStopReason Failure("Next state %s does not exist".format(nextState.stateName)))} else {nextState.replies.reverse foreach { r => sender ! r }if (currentState.stateName != nextState.stateName) {this.nextState = nextStatehandleTransition(currentState.stateName, nextState.stateName)gossip(Transition(self, currentState.stateName, nextState.stateName))this.nextState = null}currentState = nextState...}}

makeTransition 先检查目标状态是否存在(614行)，不存在则启动状态机终止流程。然后用当前状态和目标状态的元组做參数调用 handleTransition 方法。

    private def handleTransition(prev: S, next: S) {val tuple = (prev, next)for (te ← transitionEvent) { if (te.isDefinedAt(tuple)) te(tuple) }} 

handleTransition 从全部保存的状态处理函数中找到匹配的进行调用(538行)。从这里能够看出来，假设我们把onTransition 函数调用了两次。每次给的參数是一样的（都是从状态A转换到状态B）。那么这两次注冊的转换处理函数都会被调用。顺序就是依照注冊的顺序。

makeTransition 函数会在状态转换函数被运行完毕之后才将当前状态设置为目标状态(624行)。

processEvent 中假设发现当前状态的事件处理函数不能处理某个消息会调用 handleEvent(event) 函数。

  private val handleEventDefault: StateFunction = {case Event(value, stateData) ⇒log.warning("unhandled event " + value + " in state " + stateName)stay}private var handleEvent: StateFunction = handleEventDefault

handleEvent 是一个可变变量初始值被设置为 handleEventDefault ，缺省实现是输出一行日志。然后保留 在当前状态。我们能够用 whenUnhandled （DSL）来改变这个缺省行为。

    final def whenUnhandled(stateFunction: StateFunction): Unit =handleEvent = stateFunction orElse handleEventDefault

whenUnhandled 提供一个未知事件处理函数，并把这个函数置于handleEventDefault 之前被调用，假设我们提供的未知事件处理函数被调用了。缺省的函数就不会被调用。

状态事件处理函数要求我们返回下一个状态实例，这里是提出状态转换要求的唯一地方。

你能提出的要求非常easy。要么留下（stay）。要么走人(goto)。stay 事实上是goto 到当前状态。

    /*** Produce transition to other state. Return this from a state function in* order to effect the transition.** @param nextStateName state designator for the next state* @return state transition descriptor*/final def goto(nextStateName: S): State = FSM.State(nextStateName, currentState.stateData)/*** Produce "empty" transition descriptor. Return this from a state function* when no state change is to be effected.** @return descriptor for staying in current state*/final def stay(): State = goto(currentState.stateName) // cannot directly use currentState because of the timeout field

Akka FSM 的状态超时指的是“进入某个状态后一定时间内没有收到不论什么事件”。

       if (timeout.isDefined) {val t = timeout.getif (t.isFinite && t.length >= 0) {import context.dispatchertimeoutFuture = Some(context.system.scheduler.scheduleOnce(t, self, TimeoutMarker(generation)))}}

在FSM的伴生对象中定义了定时器类型。自带了调度和取消的函数。

    private[akka] case class Timer(name: String, msg: Any, repeat: Boolean, generation: Int)(context: ActorContext)extends NoSerializationVerificationNeeded {private var ref: Option[Cancellable] = _ //用于取消定时调度的句柄private val scheduler = context.system.schedulerprivate implicit val executionContext = context.dispatcherdef schedule(actor: ActorRef, timeout: FiniteDuration): Unit =ref = Some(if (repeat) scheduler.schedule(timeout, timeout, actor, this)else scheduler.scheduleOnce(timeout, actor, this))def cancel(): Unit =if (ref.isDefined) {ref.get.cancel()ref = None}}

定时时间调度使用的是 ActorSystem 中的调度器。每一个定时器都有自己的名称 (name)。schedule 函数用来运行调度。并保存用于取消的句柄。

99~100行调用ActorSystem 的调度器设定给Actor 发送消息的时间，注意，发送的消息就是 Timer 对象自己。

FSM Trait 中定义了一个Map用来依据名字保存定时器。

    private val timers = mutable.Map[String, Timer]()

用于外部调用的 API 是setTimer 函数 和 cancelTimer函数。

setTimer 构造Timer实例。保存在 timers Map中 通过Timer的schedule方法设定让Actor在一定时间后收到消息，消息内容就是Timer对象自己。消息的接收者就是当前FSM的Actor。

    final def setTimer(name: String, msg: Any, timeout: FiniteDuration, repeat: Boolean): Unit = {if (debugEvent)log.debug("setting " + (if (repeat) "repeating " else "") + "timer '" + name + "'/" + timeout + ": " + msg)if (timers contains name) {timers(name).cancel}val timer = Timer(name, msg, repeat, timerGen.next)(context)timer.schedule(self, timeout)timers(name) = timer}/*** Cancel named timer, ensuring that the message is not subsequently delivered (no race).* @param name of the timer to cancel*/final def cancelTimer(name: String): Unit = {if (debugEvent)log.debug("canceling timer '" + name + "'")if (timers contains name) {timers(name).canceltimers -= name}}    

FSM Trait 的 receive函数收到 Timer 消息后，取出Timer中的业务事件，调用状态机的事件处理函数。定时器处理会取消装的超时机制。

    case t @ Timer(name, msg, repeat, gen) =>if ((timers contains name) && (timers(name).generation == gen)) {if (timeoutFuture.isDefined) {timeoutFuture.get.cancel()timeoutFuture = None}generation += 1if (!repeat) {timers -= name}processMsg(msg, t)}

私有的 terminate 函数会启动终止流程。搜索这个函数出现的地方就能够发现状态机进入终止流程的原因：

来看看 terminate 函数：

  private def terminate(nextState: State): Unit = {if (currentState.stopReason.isEmpty) { //当前状态必须没要求终止val reason = nextState.stopReason.get //下一个状态必须有终止原因logTermination(reason)for (timer ← timers.values) timer.cancel() //取消全部定时器timers.clear()currentState = nextStateval stopEvent = StopEvent(reason, currentState.stateName, currentState.stateData)if (terminateEvent.isDefinedAt(stopEvent))terminateEvent(stopEvent) //再给用户一个机会来处理状态机终止事件}}

终止流程会取消全部的定时器(657行)。同一时候构造一个 stopEvent 事件。让用户可以对终止进行最后的处理。

要处理 stopEvent。你须要使用onTermination DSL 语法注冊一个终止事件处理函数。

    final def onTermination(terminationHandler: PartialFunction[StopEvent, Unit]): Unit =terminateEvent = terminationHandler

    private var terminateEvent: PartialFunction[StopEvent, Unit] = NullFunction

从使用者角度看。有两个地方能够对终止流程进行控制：

假设是因为 Actor 终止（出现异常或正常退出Actor)，postStop函数会以当前状态启动终止流程(650行)。

FSM.State 的using 方法是唯一能够改变状态数据的函数 在状态的事件处理函数中，收到事件后，通过currentState 获取当前的状态数据，构造下一个状态的 FSM.State实例， 假设须要改变数据就把数据放进去。

不须要改变就 goto 或 stay。

在转换处理函数中 (onTransition) ,能够通过 stateData / nextStateData 活动当前状态和下一个状态的数据。记住， 转换出炉函数运行完毕之后。状态才会被切换到下一个状态。

可是在转换处理函数中是无法改动状态数据的。当然你能够通过 nextStateData 得到数据，然后改动当中的 var 变量。可是最好别这么干，你的状态数据假设都定义成 val 的，那就完美了。

状态实例 FSM.Stat[S,D]中有一个 replies 參数，类型是 List[Any]，也提供了一个方法操作设个数据：

    def replying(replyValue: Any): State[S, D] = {copy(replies = replyValue :: replies)}

replying 会返回一个新的状态实例将提供的參数replyValue增加到新实例的replies列表中。

在状态转换时。在前一个状态的事件处理函数被调用之后，转换处理函数被调用之前。会给消息的发送者(sender )回复下一个状态实例的replies中保存的全部内容(makeTransition。617行)。终止流程启动前也会做这件事情（applyState。607行）。

由上面的分析可见。reply机制能够用来在状态转换前向消息的发送者回复不论什么信息。指定信息的方式是在状态的事件处理函数返回下一个状态时将要回复的信息保存在下一个状态的 replies 列表中。

Akka FSM 提供了两套消息来供外部订阅状态的转换。

订阅和取消都是操作 akka.routing.Listeners 中的 listeners: Set[ActorRef] 成员。状态转换时。在转换处理函数运行完毕之后会给监听者发送转换通知(makeTransition，621行)。

    gossip(Transition(self, currentState.stateName, nextState.stateName))

    protected def gossip(msg: Any)(implicit sender: ActorRef = Actor.noSender): Unit = {val i = listeners.iteratorwhile (i.hasNext) i.next ! msg}

假设不直接混入FSM Trait 而是混入 LoggingFSM Trait 就能够启用日志功能。

还要把配置文件里的 akka.actor.debug.fsm 设置为 true ，以启用日志。

LoggingFSM 重载 了 FSM 的processEvent 方法，会在每一次状态转换时输出日志。

LoggingFSM 还提供了一个内存中的滚动日志（rolling log），你能够指定一个日志队列的长度，最新的日志会填入到日志，满了就覆盖旧的。滚动日志通过 getLog 函数来获得。