FSM

    示例项目

    你可以查看「」，以了解实际应用中的情况。

    FSM（有限状态机）是一个抽象的基类，它实现了一个 Akka Actor，并在「Erlang设 计原则」中得到了最好的描述。

    FSM 可以描述为一组形式的关系：

    • State(S) x Event(E) -> Actions (A), State(S’)

    这些关系被解释为如下含义：

    • 如果我们处于状态S，并且事件E发生，那么我们应该执行操作A，并向状态S’过渡。

    一个简单的例子

    为了演示AbstractFSM类的大部分特性，考虑一个 Actor，该 Actor 在消息到达突发（burst）时接收和排队消息，并在突发结束或收到刷新（flush）请求后发送它们。

    首先，考虑使用以下所有导入语句：

    1. import akka.actor.AbstractFSM;
    2. import akka.actor.ActorRef;
    3. import akka.japi.pf.UnitMatch;
    4. import java.util.Arrays;
    5. import java.util.LinkedList;
    6. import java.util.List;
    7. import java.time.Duration;

    我们的“Buncher” Actor 的协议（contract）是接受或产生以下信息：

    1. static final class SetTarget {
    2.   private final ActorRef ref;
    4.   public SetTarget(ActorRef ref) {
    5.     this.ref = ref;
    6.   }
    8.   public ActorRef getRef() {
    9.     return ref;
    10.   }
    12.   @Override
    13.   public String toString() {
    14.     return "SetTarget{" + "ref=" + ref + '}';
    15.   }
    16. }
    18. static final class Queue {
    19.   private final Object obj;
    21.   public Queue(Object obj) {
    22.     this.obj = obj;
    23.   }
    25.   public Object getObj() {
    26.     return obj;
    27.   }
    29.   @Override
    30.   public String toString() {
    31.     return "Queue{" + "obj=" + obj + '}';
    32.   }
    33. }
    35. static final class Batch {
    36.   private final List<Object> list;
    38.   public Batch(List<Object> list) {
    39.     this.list = list;
    40.   }
    42.   public List<Object> getList() {
    43.     return list;
    44.   }
    46.   @Override
    47.   public boolean equals(Object o) {
    48.     if (this == o) return true;
    49.     if (o == null || getClass() != o.getClass()) return false;
    51.     Batch batch = (Batch) o;
    53.     return list.equals(batch.list);
    54.   }
    56.   @Override
    57.   public int hashCode() {
    58.     return list.hashCode();
    59.   }
    61.   @Override
    62.   public String toString() {
    63.     final StringBuilder builder = new StringBuilder();
    64.     builder.append("Batch{list=");
    65.     list.stream()
    66.         .forEachOrdered(
    67.             e -> {
    68.               builder.append(e);
    69.               builder.append(",");
    70.             });
    71.     int len = builder.length();
    72.     builder.replace(len, len, "}");
    73.     return builder.toString();
    74.   }
    75. }
    77. static enum Flush {
    78.   Flush
    79. }

    启动它需要SetTarget，为要传递的Batches设置目标；Queue将添加到内部队列，而Flush将标记突发（burst）的结束。

    1. // states
    2. enum State {
    3.   Idle,
    4.   Active
    5. }
    7. // state data
    8. interface Data {}
    10. enum Uninitialized implements Data {
    11.   Uninitialized
    12. }
    14. final class Todo implements Data {
    15.   private final ActorRef target;
    16.   private final List<Object> queue;
    18.   public Todo(ActorRef target, List<Object> queue) {
    19.     this.target = target;
    20.     this.queue = queue;
    21.   }
    23.   public ActorRef getTarget() {
    24.     return target;
    25.   }
    27.   public List<Object> getQueue() {
    28.     return queue;
    29.   }
    31.   @Override
    32.     return "Todo{" + "target=" + target + ", queue=" + queue + '}';
    33.   }
    35.   public Todo addElement(Object element) {
    36.     List<Object> nQueue = new LinkedList<>(queue);
    37.     nQueue.add(element);
    38.     return new Todo(this.target, nQueue);
    39.   }
    41.   public Todo copy(List<Object> queue) {
    42.     return new Todo(this.target, queue);
    43.   }
    45.   public Todo copy(ActorRef target) {
    46.     return new Todo(target, this.queue);
    47.   }
    48. }

    Actor 可以处于两种状态：没有消息排队（即Idle）或有消息排队（即Active）。它将保持Active状态，只要消息一直到达并且不请求刷新。Actor 的内部状态数据由发送的目标 Actor 引用和消息的实际队列组成。

    现在让我们来看看我们的FSM Actor 的结构（skeleton）：

    1. public class Buncher extends AbstractFSM<State, Data> {
    2.   {
    3.     startWith(Idle, Uninitialized);
    5.     when(
    6.         Idle,
    7.         matchEvent(
    8.             SetTarget.class,
    9.             Uninitialized.class,
    10.             (setTarget, uninitialized) ->
    11.                 stay().using(new Todo(setTarget.getRef(), new LinkedList<>()))));
    13.     onTransition(
    14.         matchState(
    15.                 Active,
    16.                 Idle,
    17.                 () -> {
    18.                   // reuse this matcher
    20.                       UnitMatch.create(
    21.                           matchData(
    22.                               Todo.class,
    23.                               todo ->
    24.                                   todo.getTarget().tell(new Batch(todo.getQueue()), getSelf())));
    25.                   m.match(stateData());
    26.                 })
    27.             .state(
    28.                 Idle,
    29.                 Active,
    30.                 () -> {
    31.                   /* Do something here */
    32.                 }));
    34.     when(
    35.         Active,
    36.         Duration.ofSeconds(1L),
    37.         matchEvent(
    38.             Arrays.asList(Flush.class, StateTimeout()),
    39.             Todo.class,
    40.             (event, todo) -> goTo(Idle).using(todo.copy(new LinkedList<>()))));
    42.     whenUnhandled(
    43.         matchEvent(
    44.                 Queue.class,
    45.                 Todo.class,
    46.                 (queue, todo) -> goTo(Active).using(todo.addElement(queue.getObj())))
    47.             .anyEvent(
    48.                 (event, state) -> {
    49.                   log()
    50.                       .warning(
    51.                           "received unhandled request {} in state {}/{}",
    52.                           event,
    53.                           stateName(),
    54.                           state);
    55.                   return stay();
    56.                 }));
    58.     initialize();
    59.   }
    60. }

    基本策略是通过继承AbstractFSM类并将可能的状态和数据值指定为类型参数来声明 Actor。在 Actor 的主体中，DSL 用于声明状态机：

    • startWith定义初始状态和初始数据
    • when(<state>) { ... }是要处理的每个状态的声明（可能是多个状态，传递的PartialFunction将使用orElse连接）
    • 最后使用initialize启动它，它执行到初始状态的转换并设置定时器（如果需要）。

    在这种情况下，我们从Idle状态开始，使用Uninitialized数据，其中只处理SetTarget()消息；stay准备结束此事件的处理，以避免离开当前状态，而using修饰符使 FSM 用包含目标 Actor 引用的Todo()对象替换内部状态（此时Uninitialized在这个点）。Active状态已声明状态超时，这意味着如果在 1 秒内没有收到消息，将生成FSM.StateTimeout 消息。这与在这种情况下接收Flush命令的效果相同，即转换回Idle状态并将内部队列重置为空向量。但是消息是如何排队的呢？由于这两种状态下的工作方式相同，因此我们利用以下事实：未由when()块处理的任何事件都传递给whenUnhandled()块：

    1. whenUnhandled(
    2.     matchEvent(
    3.             Queue.class,
    4.             Todo.class,
    5.             (queue, todo) -> goTo(Active).using(todo.addElement(queue.getObj())))
    6.         .anyEvent(
    7.             (event, state) -> {
    8.               log()
    9.                   .warning(
    10.                       "received unhandled request {} in state {}/{}",
    11.                       event,
    12.                       stateName(),
    13.                       state);
    14.               return stay();
    15.             }));

    这里处理的第一个案例是将Queue() 请求添加到内部队列并进入Active状态（如果已经存在的话，这显然会保持Active状态），但前提是在接收到Queue()事件时，FSM 数据没有Uninitialized。否则，在所有其他未处理的情况下，第二种情况只会记录一个警告，而不会更改内部状态。

    唯一缺少的部分是Batches实际发送到目标的位置，为此我们使用了onTransition机制：你可以声明多个这样的块，如果发生状态转换（即只有当状态实际更改时），所有这些块都将尝试匹配行为。

    1. onTransition(
    2.     matchState(
    3.             Active,
    4.             Idle,
    5.             () -> {
    6.               // reuse this matcher
    7.               final UnitMatch<Data> m =
    8.                   UnitMatch.create(
    9.                       matchData(
    10.                           Todo.class,
    11.                           todo ->
    12.                               todo.getTarget().tell(new Batch(todo.getQueue()), getSelf())));
    13.               m.match(stateData());
    14.             })
    15.         .state(
    16.             Idle,
    17.             Active,
    18.             () -> {
    19.               /* Do something here */
    20.             }));

    转换回调是由matchState构造的一个生成器，后跟零或多个state，它将当前state和下一个state作为一对状态的输入。在状态更改期间，旧的状态数据通过stateData()可用，如展示的这样，新的状态数据将作为nextStateData()可用。

    • 注释：可以使用goto(S)或stay()实现相同的状态转换（当前处于状态S时）。不同之处在于，goto(S)会发出一个事件S->S，该事件可以由onTransition处理，而stay()则不会。

    为了验证这个Buncher是否真的有效，使用「TestKit」编写一个测试非常容易，这里使用 JUnit 作为示例：

    1. public class BuncherTest extends AbstractJavaTest {
    3.   static ActorSystem system;
    5.   @BeforeClass
    6.   public static void setup() {
    7.     system = ActorSystem.create("BuncherTest");
    8.   }
    10.   @AfterClass
    11.   public static void tearDown() {
    12.     TestKit.shutdownActorSystem(system);
    13.     system = null;
    14.   }
    16.   @Test
    17.   public void testBuncherActorBatchesCorrectly() {
    18.     new TestKit(system) {
    19.       {
    20.         final ActorRef buncher = system.actorOf(Props.create(Buncher.class));
    21.         final ActorRef probe = getRef();
    23.         buncher.tell(new SetTarget(probe), probe);
    24.         buncher.tell(new Queue(43), probe);
    25.         LinkedList<Object> list1 = new LinkedList<>();
    26.         list1.add(42);
    27.         list1.add(43);
    28.         expectMsgEquals(new Batch(list1));
    29.         buncher.tell(new Queue(44), probe);
    30.         buncher.tell(Flush, probe);
    31.         buncher.tell(new Queue(45), probe);
    32.         LinkedList<Object> list2 = new LinkedList<>();
    33.         list2.add(44);
    34.         expectMsgEquals(new Batch(list2));
    35.         LinkedList<Object> list3 = new LinkedList<>();
    36.         list3.add(45);
    37.         expectMsgEquals(new Batch(list3));
    38.         system.stop(buncher);
    39.       }
    40.     };
    41.   }
    43.   @Test
    44.   public void testBuncherActorDoesntBatchUninitialized() {
    45.     new TestKit(system) {
    46.       {
    47.         final ActorRef buncher = system.actorOf(Props.create(Buncher.class));
    48.         final ActorRef probe = getRef();
    50.         buncher.tell(new Queue(42), probe);
    51.         expectNoMessage();
    52.         system.stop(buncher);
    53.       }
    54.     };
    55.   }
    56. }

    抽象类是用于实现 FSM 的基类。它实现了 Actor，因为创建了一个 Actor 来驱动 FSM。

    • 注释：AbstractFSM类定义了一个receive方法，该方法处理内部消息，并将其他所有信息传递给 FSM 逻辑（根据当前状态）。当覆盖receive方法时，请记住，例如状态超时处理取决于通过 FSM 逻辑实际传递消息。

    AbstractFSM类采用两个类型参数：

    • 所有状态名的父类型，通常是枚举
    • AbstractFSM模块本身跟踪的状态数据的类型。

    定义状态

    状态由方法的一个或多个调用定义。

    • when(<name>[, stateTimeout = <timeout>])(stateFunction)

    给定的名称必须是与AbstractFSM类的第一个类型参数类型兼容的对象。此对象用作哈希键，因此必须确保它正确实现equals和hashCode；尤其是它不能是可变的。最适合这些需求的是case对象。

    如果给定stateTimeout参数，那么默认情况下，所有转换到该状态（包括保持）的操作都将接收该超时。使用显式超时启动转换可用于重写此默认值，有关详细信息，请参阅「Initiating Transitions」。在使用setStateTimeout(state, duration)进行操作处理期间，可以更改任何状态的状态超时。这将启用运行时配置，例如通过外部消息。

    stateFunction参数是一个PartialFunction[Event, State]，它使用状态函数生成器语法方便地给出，如下所示：

    1. when(
    2.     Idle,
    3.     matchEvent(
    4.         SetTarget.class,
    5.         Uninitialized.class,
    6.         (setTarget, uninitialized) ->
    7.             stay().using(new Todo(setTarget.getRef(), new LinkedList<>()))));
    • 警告：需要为每个可能的 FSM 状态定义处理程序，否则在尝试切换到未声明的状态时将出现故障。

    建议将状态声明为枚举，然后验证每个状态都有一个when子句。如果要使状态的处理“unhandled”（下面将详细介绍），则仍需要这样声明：

    1. when(SomeState, AbstractFSM.NullFunction());

    定义初始状态

    每个 FSM 都需要一个起点（starting point），该起点使用：

    1. startWith(state, data[, timeout])

    可选的给定超时参数重写为所需初始状态给定的任何规范。如果要取消默认超时，请使用Duration.Inf。

    未处理的事件

    如果状态不处理接收到的事件，则会记录警告。如果要在这种情况下执行其他操作，可以使用whenUnhandled(stateFunction)指定：

    1.  whenUnhandled(
    2.       matchEvent(
    3.               X.class,
    4.               (x, data) -> {
    5.                 log().info("Received unhandled event: " + x);
    6.                 return stay();
    7.               })
    8.           .anyEvent(
    9.               (event, data) -> {
    10.                 log().warning("Received unknown event: " + event);
    11.                 return goTo(Error);
    12.               }));
    13. }

    在此处理程序中，可以使用stateName方法查询 FSM 的状态。

    • 重要的：此处理程序不是堆叠的，这意味着每次调用whenUnhandled都会替换先前安装的（installed）处理程序。

    任何stateFunction的结果都必须是下一个状态的定义，除非终止 FSM，如「」。状态定义可以是当前状态（如stay指令所述），也可以是goto(state)给出的不同状态。结果对象允许通过下面描述的修饰符进一步限定：

    • forMax(duration)，此修饰符设置下一个状态的状态超时。这意味着计时器（timer）启动，到期时向 FSM 发送StateTimeout消息。此计时器在同时接收到任何其他消息时被取消；你可以依赖这样一个事实，即在干预消息之后将不会处理StateTimeout消息。此修饰符还可用于重写为目标状态指定的任何默认超时。如果要取消默认超时，请使用Duration.Inf。
    • using(data)，此修饰符将旧状态数据替换为给定的新数据。如果你遵循上面的建议，这是唯一一个修改内部状态数据的地方。
    • replying(msg)，此修饰符向当前处理的消息发送答复，否则不会修改状态转换。

    所有修饰符都可以链接起来，以实现一个漂亮简洁的描述：

    1. when(
    2.     SomeState,
    3.     matchAnyEvent(
    4.         (msg, data) -> {
    5.           return goTo(Processing)
    6.               .using(newData)
    7.               .forMax(Duration.ofSeconds(5))
    8.               .replying(WillDo);
    9.         }));

    实际上并非所有情况下都需要括号，但它们在视觉上区分修饰符和它们的参数，因此使代码更易于阅读。

    • 注释：请注意，return语句不能在when块或类似块中使用；这是一个 Scala 限制。使用if () ... else ...或者将其移动到方法定义中。

    监视转换

    概念上，“状态之间”会发生转换，这意味着在将任何操作放入事件处理块之后，这是显而易见的，因为下一个状态仅由事件处理逻辑返回的值定义。你不必担心设置内部状态变量的确切顺序，因为 FSM Actor 中的所有内容都在以单线程运行。

    内部监控

    到目前为止，FSM DSL 一直以状态和事件为中心。双视图（dual view）将其描述为一系列转换。这是由方法启用的

    1. onTransition(handler)

    它将动作与转换相关联，而不是与状态和事件相关联。处理程序是一个以一对状态作为输入的部分函数；不需要结果状态，因为无法修改正在进行的转换。

    1. onTransition(
    2.     matchState(Idle, Active, () -> setTimer("timeout", Tick, Duration.ofSeconds(1L), true))
    3.         .state(Active, null, () -> cancelTimer("timeout"))
    4.         .state(null, Idle, (f, t) -> log().info("entering Idle from " + f)));

    也可以将接受两种状态的函数对象传递给onTransition，以将转换处理逻辑实现为一种方法：

    使用此方法注册的处理程序是堆叠（stacked）的，因此你可以在适合你的设计块中散置intersperse块。但是，应该注意的是，要为每个转换（transition）调用所有处理程序，而不仅仅是第一个匹配的处理程序。这是专门设计的，这样你就可以将某个方面的所有转换处理放在一个地方，而不必担心前面的声明会影响后面的声明；不过，操作仍然是按声明顺序执行的。

    • 注释：这种内部监控可用于根据转换构造你的 FSM，例如，在添加新的目标状态时，不能忘记在离开某个状态时取消计时器。

    外部监控

    外部 Actor 可以通过发送消息SubscribeTransitionCallBack(actorRef)来注册以获得状态转换的通知。命名的 Actor 将立即发送一条CurrentState(self, stateName)消息，并在触发状态更改时接收Transition(actorRef, oldState, newState)消息。

    在不注销的情况下停止侦听器（listener）将不会从订阅列表中删除该侦听器；请在停止侦听器之前使用UnsubscribeTransitionCallback。

    除了状态超时之外，FSM 还管理由String名称标识的定时器（timers）。你可以使用

    1. setTimer(name, msg, interval, repeat)

    其中msg是将在持续时间interval结束后发送的消息对象。如果repeat为true，则计时器按interval参数给定的固定速率调度。在添加新计时器之前，任何具有相同名称的现有计时器都将自动取消。

    计时器取消可以使用：

    1. cancelTimer(name)

    它保证立即工作，这意味着即使计时器已经启动并将其排队，也不会在调用后处理计划的消息。任何计时器的状态都可以通过以下方式获取：

    1. isTimerActive(name)

    这些命名的计时器补充状态超时，因为它们不受接收其他消息的影响。

    从内部终止

    通过将结果状态指定为以下方式来停止 FSM：

    1. stop([reason[, data]])

    原因必须是Normal（默认）、Shutdown或Failure(reason)之一，并且可以给出第二个参数来更改终止处理期间可用的状态数据。

    • 注释：应该注意的是，停止不会中止动作，并立即停止 FSM。停止操作必须以与状态转换相同的方式从事件处理程序返回，但请注意，在when块中不能使用return语句。
    1. when(
    2.     Error,
    3.     matchEventEquals(
    4.         "stop",
    5.         (event, data) -> {
    6.           // do cleanup ...
    7.           return stop();
    8.         }));

    可以使用onTermination(handler)指定在 FSM 停止时执行的自定义代码。处理程序是一个分部函数，它以StopEvent(reason, stateName, stateData) 作为参数：

    1. onTermination(
    2.     matchStop(
    3.             Normal(),
    4.             (state, data) -> {
    5.               /* Do something here */
    6.             })
    7.         .stop(
    8.             Shutdown(),
    9.             (state, data) -> {
    10.               /* Do something here */
    11.             })
    12.         .stop(
    13.             Failure.class,
    14.             (reason, state, data) -> {
    15.               /* Do something here */
    16.             }));

    对于whenUnhandled案例，此处理程序不堆叠，因此每次调用onTermination都会替换先前安装的处理程序。

    从外部终止

    当使用stop()方法停止与 FSM 关联的ActorRef时，将执行其postStop钩子。AbstractFSM类的默认实现是在准备处理StopEvent(Shutdown, ...)时执行onTermination处理程序。

    • 警告：如果你重写postStop并希望调用onTermination处理程序，请不要忘记调用super.postStop。

    有限状态机的测试和调试

    在开发和故障排除过程中，FSM 和其他 Actor 一样需要关注。如「TestFSMRef」和以下所述，有专门的工具可用。

    事件跟踪

    在「配置」中设置akka.actor.debug.fsm可以通过LoggingFSM实例记录事件跟踪：

    1. static class MyFSM extends AbstractLoggingFSM<StateType, Data> {
    2.   @Override
    3.   public int logDepth() {
    4.     return 12;
    5.   }
    7.   {
    8.     onTermination(
    9.         matchStop(
    10.             Failure.class,
    11.             (reason, state, data) -> {
    12.               String lastEvents = getLog().mkString("\n\t");
    13.               log()
    14.                   .warning(
    15.                       "Failure in state "
    16.                           + state
    17.                           + " with data "
    18.                           + data
    19.                           + "\n"
    20.                           + "Events leading up to this point:\n\t"
    21.                           + lastEvents);
    22.             }));
    23.     // ...
    24.   }
    25. }

    此 FSM 将在DEBUG级别记录日志：

    • 所有已处理的事件，包括StateTimeout和定时计时器消息
    • 每次设置和取消指定计时器
    • 所有状态转换

    生命周期更改和特殊消息可以按照对「」的描述进行记录。

    AbstractLoggingFSM类向 FSM 添加了另一个功能：滚动事件日志（rolling event log），可在调试期间（用于跟踪 FSM 如何进入特定故障状态）或其他创造性用途中使用：

    logDepth默认为零，这将关闭事件日志。

    • 警告：日志缓冲区是在 Actor 创建期间分配的，这就是使用虚拟方法调用完成配置的原因。如果要使用val进行重写，请确保其初始化发生在运行LoggingFSM的初始值设定项之前，并且不要在分配缓冲区后更改logDepth返回的值。

    事件日志的内容可使用getLog方法获取，该方法返回IndexedSeq[LogEntry]，其中最早的条目位于索引零。

    英文原文链接：FSM.