文档章节

Akka源码分析-FSM

o
 osc_4nmshwhm
发布于 2018/08/06 16:28
字数 2845
阅读 0
收藏 0

精选30+云产品,助力企业轻松上云!>>>

  akka还有一个不常使用、但我觉得比较方便的一个模块,那就是FSM(有限状态机)。我们知道了akka中Actor模型的具体实现之后,就会发现,Akka的actor可以非常方便的实现FSM。其实在akka中实现FSM还是非常简单的,或者说我们在akka中有意无意的都在使用FSM。

  在介绍分析akka中FSM源码之前,先简单介绍下FSM基本概念。一个有限状态机是一个设备,或者是一个设备模型,具有有限数量的状态,它可以在任何给定的时间根据输入进行操作,使得一个状态变换到另一个状态,或者是使一个输入或者一种行为的发生。一个有限状态机在任何瞬间只能处在一种状态。FSM有几个重要的概念:状态、行为、状态变换条件。状态就是当前FSM所处的状态,行为就是当前状态下FSM对输入的响应,状态变换就是指FSM在什么条件下需要更改当前的状态。

  那么如何将FSM映射到akka中呢?其实非常简单。FSM就是一个actor;状态就是actor中的某个变量的值;行为其实可以定义为某个状态下的receive函数;状态转移,定义成某个函数就行了,只不过什么时候转移需要开发者自定义。当然这只是我们在没有分析源码之前的想当然,那具体是怎么样的呢?

/**
   * This captures all of the managed state of the [[akka.actor.FSM]]: the state
   * name, the state data, possibly custom timeout, stop reason and replies
   * accumulated while processing the last message.
   */
  case class State[S, D](stateName: S, stateData: D, timeout: Option[FiniteDuration] = None, stopReason: Option[Reason] = None, replies: List[Any] = Nil)

   上面是状态的定义,有两个类型参数:S、D。分别是状态名称和状态数据,其中状态名称比较容易理解,状态数据可以认为是与当前状态关联的值。timeout标志当前状态持续的最长时间,这一点比较特殊,普通的状态机,如果没有主动更改状态会一直持续在当前状态的。stopReason当前状态停止的原因。replies暂时不做分析,可以理解为一种通知机制,退出当前状态时对listener的通知。

/**
   * All messages sent to the [[akka.actor.FSM]] will be wrapped inside an
   * `Event`, which allows pattern matching to extract both state and data.
   */
  final case class Event[D](event: Any, stateData: D) extends NoSerializationVerificationNeeded

   FSM还定义了一个Event,用来对用户数据的封装,用来做模式匹配。

type StateFunction = scala.PartialFunction[Event, State]

   上面是FSM行为的定义,比较简单,就是一个PartialFunction,它输入一个Event,输出下一个状态。其实从这一点也可以看出,转义状态是需要开发手动指定的。

/*
* A fsm hakker is an awesome dude or dudette who either thinks about hacking or has to eat ;-)
*/
class FSMHakker(name: String, left: ActorRef, right: ActorRef) extends Actor with FSM[FSMHakkerState, TakenChopsticks] {

  //All hakkers start waiting
  startWith(Waiting, TakenChopsticks(None, None))

  when(Waiting) {
    case Event(Think, _) =>
      println("%s starts to think".format(name))
      startThinking(5.seconds)
  }

  //When a hakker is thinking it can become hungry
  //and try to pick up its chopsticks and eat
  when(Thinking) {
    case Event(StateTimeout, _) =>
      left ! Take
      right ! Take
      goto(Hungry)
  }

  // When a hakker is hungry it tries to pick up its chopsticks and eat
  // When it picks one up, it goes into wait for the other
  // If the hakkers first attempt at grabbing a chopstick fails,
  // it starts to wait for the response of the other grab
  when(Hungry) {
    case Event(Taken(`left`), _) =>
      goto(WaitForOtherChopstick) using TakenChopsticks(Some(left), None)
    case Event(Taken(`right`), _) =>
      goto(WaitForOtherChopstick) using TakenChopsticks(None, Some(right))
    case Event(Busy(_), _) =>
      goto(FirstChopstickDenied)
  }

  // When a hakker is waiting for the last chopstick it can either obtain it
  // and start eating, or the other chopstick was busy, and the hakker goes
  // back to think about how he should obtain his chopsticks :-)
  when(WaitForOtherChopstick) {
    case Event(Taken(`left`), TakenChopsticks(None, Some(right))) => startEating(left, right)
    case Event(Taken(`right`), TakenChopsticks(Some(left), None)) => startEating(left, right)
    case Event(Busy(chopstick), TakenChopsticks(leftOption, rightOption)) =>
      leftOption.foreach(_ ! Put)
      rightOption.foreach(_ ! Put)
      startThinking(10.milliseconds)
  }

  private def startEating(left: ActorRef, right: ActorRef): State = {
    println("%s has picked up %s and %s and starts to eat".format(name, left.path.name, right.path.name))
    goto(Eating) using TakenChopsticks(Some(left), Some(right)) forMax (5.seconds)
  }

  // When the results of the other grab comes back,
  // he needs to put it back if he got the other one.
  // Then go back and think and try to grab the chopsticks again
  when(FirstChopstickDenied) {
    case Event(Taken(secondChopstick), _) =>
      secondChopstick ! Put
      startThinking(10.milliseconds)
    case Event(Busy(chopstick), _) =>
      startThinking(10.milliseconds)
  }

  // When a hakker is eating, he can decide to start to think,
  // then he puts down his chopsticks and starts to think
  when(Eating) {
    case Event(StateTimeout, _) =>
      println("%s puts down his chopsticks and starts to think".format(name))
      left ! Put
      right ! Put
      startThinking(5.seconds)
  }

  // Initialize the hakker
  initialize()

  private def startThinking(duration: FiniteDuration): State = {
    goto(Thinking) using TakenChopsticks(None, None) forMax duration
  }
}

   上面是官方的一个demo,有几个重要的点需要注意。startWith用来定义状态机起始状态;when用来定义某个状态下对Event的处理行为;goto用来转移状态;using用来定义转移状态时,当前状态关联的值;initialize用来初始化。下面我们逐步分析FSM这些概念的实现源码。

/**
   * Set initial state. Call this method from the constructor before the [[#initialize]] method.
   * If different state is needed after a restart this method, followed by [[#initialize]], can
   * be used in the actor life cycle hooks [[akka.actor.Actor#preStart]] and [[akka.actor.Actor#postRestart]].
   *
   * @param stateName initial state designator
   * @param stateData initial state data
   * @param timeout state timeout for the initial state, overriding the default timeout for that state
   */
  final def startWith(stateName: S, stateData: D, timeout: Timeout = None): Unit =
    currentState = FSM.State(stateName, stateData, timeout)

 

private var currentState: State = _

   startWith实现非常简单,就是给currentState赋值,官网注释也说了,必须在initialize之前调用。挺不喜欢这种设计的,居然需要跟调用顺序绑定。

/**
   * Insert a new StateFunction at the end of the processing chain for the
   * given state. If the stateTimeout parameter is set, entering this state
   * without a differing explicit timeout setting will trigger a StateTimeout
   * event; the same is true when using #stay.
   *
   * @param stateName designator for the state
   * @param stateTimeout default state timeout for this state
   * @param stateFunction partial function describing response to input
   */
  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 function
      stateTimeouts(name) = timeout orElse stateTimeouts(name)
    } else {
      stateFunctions(name) = function
      stateTimeouts(name) = timeout
    }
  }

 

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

   综合上面三段代码,其实when方法就是把状态名与对应的行为之间的对应关系保存到一个map中,但需要注意的是,map中保存的是状态行为的一个合并,如果对同一个状态多次调用when,则每次注册的函数会依次调用,如果已经命中某个Event则不会再往后传递执行。

/**
   * Produce transition to other state.
   * Return this from a state function in order to effect the transition.
   *
   * This method always triggers transition events, even for `A -> A` transitions.
   * If you want to stay in the same state without triggering an state transition event use [[#stay]] instead.
   *
   * @param nextStateName state designator for the next state
   * @return state transition descriptor
   */
  final def goto(nextStateName: S): State = FSM.State(nextStateName, currentState.stateData)

   goto看起来非常简单,它就是构造了下一个状态,那么为啥状态就会转移了呢?这个后面再分析。官网注释里面说,调用这个方法时会触发状态转移,而且会触发转移事件,即使是同状态的转义;如果不想触发状态转移事件,则可以使用stay方法,那么stay方法是什么呢?

/**
   * Produce "empty" transition descriptor.
   * Return this from a state function when no state change is to be effected.
   *
   * No transition event will be triggered by [[#stay]].
   * If you want to trigger an event like `S -> S` for `onTransition` to handle use `goto` instead.
   *
   * @return descriptor for staying in current state
   */
  final def stay(): State = goto(currentState.stateName).withNotification(false)

   它居然还是调用了goto,不同的是调用了withNotification。

private[akka] def withNotification(notifies: Boolean): State[S, D] = {
      if (notifies)
        State(stateName, stateData, timeout, stopReason, replies)
      else
        new SilentState(stateName, stateData, timeout, stopReason, replies)
    }
  }

   而当notifies是false时,withNotification返回了一个SilentState

private[akka] class SilentState[S, D](_stateName: S, _stateData: D, _timeout: Option[FiniteDuration], _stopReason: Option[Reason], _replies: List[Any])
    extends State[S, D](_stateName, _stateData, _timeout, _stopReason, _replies) {

    /**
     * INTERNAL API
     */
    private[akka] override def notifies: Boolean = false

    override def copy(stateName: S = stateName, stateData: D = stateData, timeout: Option[FiniteDuration] = timeout, stopReason: Option[Reason] = stopReason, replies: List[Any] = replies): State[S, D] = {
      new SilentState(stateName, stateData, timeout, stopReason, replies)
    }
  }

   SilentState与State的唯一区别就是notifies的值是false。其实简单来说,goto和stay都是创建了下一个状态的值。

  分析到这里,startWith/when/StateFunction/goto基本就可以定义一个状态机了。那么FSM是如何进行状态的变换,或者说是如何运行的呢?其实吧,如果你对actor比较了解,就一定知道,这是在receive里面实现的。

override def receive: Receive = {
    case TimeoutMarker(gen) ⇒
      if (generation == gen) {
        processMsg(StateTimeout, "state timeout")
      }
    case t @ Timer(name, msg, repeat, gen, owner) ⇒
      if ((owner eq this) && (timers contains name) && (timers(name).generation == gen)) {
        if (timeoutFuture.isDefined) {
          timeoutFuture.get.cancel()
          timeoutFuture = None
        }
        generation += 1
        if (!repeat) {
          timers -= name
        }
        processMsg(msg, t)
      }
    case SubscribeTransitionCallBack(actorRef) ⇒
      // TODO Use context.watch(actor) and receive Terminated(actor) to clean up list
      listeners.add(actorRef)
      // send current state back as reference point
      actorRef ! CurrentState(self, currentState.stateName)
    case Listen(actorRef) ⇒
      // TODO Use context.watch(actor) and receive Terminated(actor) to clean up list
      listeners.add(actorRef)
      // send current state back as reference point
      actorRef ! CurrentState(self, currentState.stateName)
    case UnsubscribeTransitionCallBack(actorRef) ⇒
      listeners.remove(actorRef)
    case Deafen(actorRef) ⇒
      listeners.remove(actorRef)
    case value ⇒ {
      if (timeoutFuture.isDefined) {
        timeoutFuture.get.cancel()
        timeoutFuture = None
      }
      generation += 1
      processMsg(value, sender())
    }
  }

   我们只关注最后一段代码processMsg(value, sender())。

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

  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 defined
      handleEvent(event)
    }
    applyState(nextState)
  }

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

  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 || nextState.notifies) {
        this.nextState = nextState
        handleTransition(currentState.stateName, nextState.stateName)
        gossip(Transition(self, currentState.stateName, nextState.stateName))
        this.nextState = null
      }
      currentState = nextState

      def scheduleTimeout(d: FiniteDuration): Some[Cancellable] = {
        import context.dispatcher
        Some(context.system.scheduler.scheduleOnce(d, self, TimeoutMarker(generation)))
      }

      currentState.timeout match {
        case SomeMaxFiniteDuration                    ⇒ // effectively disable stateTimeout
        case Some(d: FiniteDuration) if d.length >= 0 ⇒ timeoutFuture = scheduleTimeout(d)
        case _ ⇒
          val timeout = stateTimeouts(currentState.stateName)
          if (timeout.isDefined) timeoutFuture = scheduleTimeout(timeout.get)
      }
    }
  }

   processMsg方法对用户输入的消息,进行了简单的包装,然后调用processEvent;processEvent根据当前状态名找到对应的状态函数,处理当前事件,状态函数返回的下一个状态(是用goto或stay返回的)传给applyState,进行状态转移。

  下一个状态没有stopReason则进行正常的状态转移,否则就停止当前状态机。makeTransition是最终处理状态转移的方法,其实也比较简单,会首先判断是否需要通知,需要则通知相关的listener;然后更新当前状态为最新的值,当然了,还会处理超时时间。

  其实分析道这里,FSM就没必要再深入研究了,非常简单。就是定义了一些DSL,用trait Actor的receive函数进行行为函数的调用和状态转移,如果你的状态不多,转移条件简单,完全没必要用FSM,自己用beconme/unbecome实现就行了。而且官方的FSM是用一个函数调用列表(或者说是一个函数指针,C++里面经常这样实现)实现的,并没有用beconme/unbecome,这一点倒让我挺意外的。当然了,如果你的场景确实是FSM,那最好还是用官方的实现,毕竟它已经把FSM可能遇到的问题都帮你处理好了。

  但有一点需要说明那就是akka的FSM对超时的处理机制,完全需要自己处理,比如你通过forMax给某个状态设置了超时时间,那么在该状态对应的行为中你需要处理Event(FSM.StateTimeout)消息,否则akka是不知道在某个状态超时之后应该转移到哪个状态的。akka所做的就是在某个状态指定时间内没有收到消息时,发送一个超时事件给你,而指定的时间内收到消息,则会取消上一次的超时设置,设置新的超时时间。

o
粉丝 0
博文 500
码字总数 0
作品 0
私信 提问
加载中
请先登录后再评论。

暂无文章

如果你失明了,你怎么编程? - How can you program if you're blind?

问题: Sight is one of the senses most programmers take for granted. 视觉是大多数程序员认为理所当然的感官之一。 Most programmers would spend hours looking at a computer monitor......

技术盛宴
52分钟前
16
0
如何删除使用Python的easy_install安装的软件包? - How do I remove packages installed with Python's easy_install?

问题: Python's easy_install makes installing new packages extremely convenient. Python的easy_install使安装新包非常方便。 However, as far as I can tell, it doesn't implement th......

fyin1314
今天
11
0
如何将逗号分隔的字符串转换为数组? - How to convert a comma separated string to an array?

问题: I have a comma separated string that I want to convert into an array, so I can loop through it. 我有一个逗号分隔的字符串,我想将其转换成数组,因此可以循环遍历它。 Is the...

富含淀粉
今天
13
0
深源恒际:担心个人身份被冒用?不存在!

本文作者:c****t 近日,苟晶被冒名顶替身份参加高考的事件在社会各界掀起广泛热议。事件调查结果公布后,舆论风向逆转,吃瓜群众认为当事人夸大其词消费了公众情绪,一边对当事人所遭遇的不...

百度开发者中心
昨天
5
0
CKEditor 5 + SpringBoot实战(三):SpringData JPA数据持久化

在本系列的文章中,我将介绍如何在Spring Boot Application中使用CKEditor编辑器。介绍的内容包括基本环境的搭建,文件上传,SpringData JPA数据持久化,CKEditor5的安装,CKEditor图片上传,...

树下魅狐
今天
9
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部