FSM
依赖
为了使用有限状态机(Finite State Machine)Actor,你需要将以下依赖添加到你的项目中:
1. <!-- Maven -->
2. <dependency>
3. <groupId>com.typesafe.akka</groupId>
4. <artifactId>akka-actor_2.12</artifactId>
5. <version>2.5.21</version>
6. </dependency>
8. <!-- Gradle -->
9. dependencies {
10. compile group: 'com.typesafe.akka', name: 'akka-actor_2.12', version: '2.5.21'
11. }
13. <!-- sbt -->
14. libraryDependencies += "com.typesafe.akka" %% "akka-actor" % "2.5.21"
示例项目
你可以查看「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. public String toString() {
33. return "Todo{" + "target=" + target + ", queue=" + queue + '}';
34. }
36. public Todo addElement(Object element) {
37. List<Object> nQueue = new LinkedList<>(queue);
38. nQueue.add(element);
39. return new Todo(this.target, nQueue);
40. }
42. public Todo copy(List<Object> queue) {
43. return new Todo(this.target, queue);
44. }
46. public Todo copy(ActorRef target) {
47. return new Todo(target, this.queue);
48. }
49. }
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
19. final UnitMatch<Data> m =
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(42), probe);
25. buncher.tell(new Queue(43), probe);
26. LinkedList<Object> list1 = new LinkedList<>();
27. list1.add(42);
28. list1.add(43);
29. expectMsgEquals(new Batch(list1));
30. buncher.tell(new Queue(44), probe);
31. buncher.tell(Flush, probe);
32. buncher.tell(new Queue(45), probe);
33. LinkedList<Object> list2 = new LinkedList<>();
34. list2.add(44);
35. expectMsgEquals(new Batch(list2));
36. LinkedList<Object> list3 = new LinkedList<>();
37. list3.add(45);
38. expectMsgEquals(new Batch(list3));
39. system.stop(buncher);
40. }
41. };
42. }
44. @Test
45. public void testBuncherActorDoesntBatchUninitialized() {
46. new TestKit(system) {
47. {
48. final ActorRef buncher = system.actorOf(Props.create(Buncher.class));
49. final ActorRef probe = getRef();
51. buncher.tell(new Queue(42), probe);
52. expectNoMessage();
53. system.stop(buncher);
54. }
55. };
56. }
57. }
引用
AbstractFSM 类
AbstractFSM抽象类是用于实现 FSM 的基类。它实现了 Actor,因为创建了一个 Actor 来驱动 FSM。
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
19. final UnitMatch<Data> m =
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类定义了一个receive方法,该方法处理内部消息,并将其他所有信息传递给 FSM 逻辑(根据当前状态)。当覆盖receive方法时,请记住,例如状态超时处理取决于通过 FSM 逻辑实际传递消息。
AbstractFSM类采用两个类型参数:
- 所有状态名的父类型,通常是枚举
AbstractFSM模块本身跟踪的状态数据的类型。
特别地,状态数据和状态名称一起描述状态机的内部状态;如果你坚持这个方案,并且不向 FSM 类添加可变字段,则可以在一些众所周知的地方显式地进行内部状态的所有更改。
定义状态
状态由方法的一个或多个调用定义。
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,如「Termination from Inside」。状态定义可以是当前状态(如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,以将转换处理逻辑实现为一种方法:
1. public void handler(StateType from, StateType to) {
2. // handle transition here
3. }
5. onTransition(this::handler);
使用此方法注册的处理程序是堆叠(stacked)的,因此你可以在适合你的设计块中散置intersperse块。但是,应该注意的是,要为每个转换(transition)调用所有处理程序,而不仅仅是第一个匹配的处理程序。这是专门设计的,这样你就可以将某个方面的所有转换处理放在一个地方,而不必担心前面的声明会影响后面的声明;不过,操作仍然是按声明顺序执行的。
- 注释:这种内部监控可用于根据转换构造你的 FSM,例如,在添加新的目标状态时,不能忘记在离开某个状态时取消计时器。
外部监控
外部 Actor 可以通过发送消息SubscribeTransitionCallBack(actorRef)来注册以获得状态转换的通知。命名的 Actor 将立即发送一条CurrentState(self, stateName)消息,并在触发状态更改时接收Transition(actorRef, oldState, newState)消息。
通过向 FSM Actor 发送UnsubscribeTransitionCallBack(actorRef),可以注销外部监控。
在不注销的情况下停止侦听器(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和定时计时器消息 - 每次设置和取消指定计时器
- 所有状态转换
生命周期更改和特殊消息可以按照对「Actors」的描述进行记录。
滚动事件日志
AbstractLoggingFSM类向 FSM 添加了另一个功能:滚动事件日志(rolling event log),可在调试期间(用于跟踪 FSM 如何进入特定故障状态)或其他创造性用途中使用:
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. }
logDepth默认为零,这将关闭事件日志。
- 警告:日志缓冲区是在 Actor 创建期间分配的,这就是使用虚拟方法调用完成配置的原因。如果要使用
val进行重写,请确保其初始化发生在运行LoggingFSM的初始值设定项之前,并且不要在分配缓冲区后更改logDepth返回的值。
事件日志的内容可使用getLog方法获取,该方法返回IndexedSeq[LogEntry],其中最早的条目位于索引零。
示例
与 Actor 的become/unbecome相比,一个更大的 FSM 示例可以下载成一个随时可以运行「Akka FSM 示例」和一个教程。此示例的源代码也可以在「Akka Samples Repository」中找到。
英文原文链接:FSM.
