0

我有一个用例，一旦请求数量达到指定值，我必须使用 akka fsm 处理请求。

sealed trait State
case object Idle extends State
case object Active extends State

sealed trait Data
case object Uninitialized extends Data
case object QuickStart extends Data
case class A(a: Int) extends Data

class RequestHandlers extends FSM[State, Data] {
  val queue = mutable.Queue[A]()
  startWith(Idle, Uninitialized)

  when(Idle) {
    case Event(_, Uninitialized) =>
      println("At Idle")
      //      self ! QuickStart
      goto(Active) using QuickStart
  }

  when(Active) {
    case Event(_, request: A) =>
      println("At Active")
      queue.take(2).map{x => println("request---  " + x.a  + "processing")
      queue.dequeue()

}

      Thread.sleep(2000L)
      goto(Active) using Uninitialized
  }


  whenUnhandled {
    case Event(update: A, QuickStart) =>
      queue += update
      if(queue.size >= 2) {
        println(s"At unhandled + ${update}" + "--" + queue)
        goto(Active) using update
      }
      else {
        println("size has not reached")
        goto(Active) using Uninitialized
      }
    case Event(update: A, Uninitialized) =>
      queue += update
      println(s"At unhandled - Uninitialised + $update")
      goto(Active) using QuickStart
  }

  initialize()

}

object demo extends App  {

  val actorSystem = ActorSystem("system")
  val actor = actorSystem.actorOf(Props(classOf[RequestHandlers]))

  val list = (1 to 10).toList
  list.foreach { abc =>

    actor ! Uninitialized
    actor ! A(abc)
    println("Sent")
  }

}

我尝试在添加请求的地方使用可变队列。在队列大小达到某个值后，即 2 同时处理这些请求。处理后，我将其出列。如果我发送 10 个请求，它将处理 8 个请求，但对于最后 2 个请求，它永远不会进入活动状态。在过渡期间，我没有得到我犯错误的地方。

任何帮助将不胜感激！

4

1 回答 1

1

我认为您正在做的最小示例如下所示：

// The only type of incoming message
case class Msg(a: Int)

// States
sealed trait State
case object Waiting extends State
case object Active extends State

// StateData is shared between states
case class StateData(queue: immutable.Queue[Msg])
object StateData {
  val empty = StateData(immutable.Queue.empty)

  def single(msg: Msg) = StateData(immutable.Queue(msg))
}


class RequestHandlers extends FSM[State, StateData] {
  val startTime = System.currentTimeMillis()

  def curTime = {
    val time = (System.currentTimeMillis() - startTime) / 1000f
    f"[$time%3.2f]"
  }

  startWith(Waiting, StateData.empty)

  onTransition {
    case Waiting -> Active =>
      //use nextStateData rather than stateData !
      nextStateData match {
        case StateData(queue) =>
          queue.foreach(x => println(s"$curTime processing ${x.a} "))
          Thread.sleep(2000L)
      }
  }

  when(Active) {
    case Event(msg: Msg, _) =>
      println(s"$curTime at Active $msg")
      // we've just processed old data
      // drop the old queue and create a new one with the new message
      goto(Waiting) using StateData.single(msg)
  }
  when(Waiting) {
    case Event(msg: Msg, StateData(oldQueue)) =>
      // add an event to the queue and check if it is time to process
      val newQueue = oldQueue :+ msg
      println(s"$curTime at Idle $msg, newQueue = $newQueue")
      if (newQueue.size == 2) {
        goto(Active) using StateData(newQueue)
      }
      else {
        stay using StateData(newQueue)
      }
  }

  initialize()
}

测试程序是

object demo extends App  {

    val actorSystem = ActorSystem("system")
    val actor = actorSystem.actorOf(Props(classOf[RequestHandlers]))

    (1 to 10).toList.foreach { i =>
      println(s"Send $i")
      actor ! Msg(i)
    }

}

的逻辑RequestHandlers是它将传入请求累积在存储在StateData对象内的队列中（该对象只有一种在两种状态之间共享的类型）。有两种状态Waiting和Active。处理实际上发生在转换Waiting->上Active。可能最棘手的一点是不要忘记当 FSM 处于Active状态时，新消息将到达并且应该通过添加到队列（或者更确切地说使用来自该消息的数据启动一个新队列）来处理。

PS好吧，这个例子可能不是那么小。实际上，您可以只有一个状态并在内部进行处理，if (newQueue.size == 2)但这将是一个非常奇怪的 FSM。

于 2018-01-14T12:35:10.477 回答