如何使用 akka fsm 制作基于尺寸的节流阀？

Question

我有一个用例，一旦请求数量达到指定值，我必须使用 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 个，它永远不会进入活动状态。我在转换时没有弄错。

任何帮助将不胜感激！

Answer 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 状态时，新消息将到达并且应该通过添加到队列中进行处理（或者更确切地说使用该消息中的数据启动一个新队列）。

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

如何使用 akka fsm 制作基于尺寸的节流阀？

How to make size based throttler using akka fsm?

scala

throttling

akka-fsm