使用FSM进行复杂订单状态管理(2)

tech

This article was last updated on <span id="expire-date"></span> days ago, the information described in the article may be outdated.

接着上一篇文章,我们这次来设计一个简单可扩展的有限状态机。

根据有限状态机的定义,我们的状态机将有三部分组成:

  • 状态类 - 负责承载状态信息
  • 事件类 - 负责记录事件发生的信息并执行转移函数
  • 管理器类 - 负责记录状态转移表并管理当前的状态

下面让我们开始建模。

状态类

我们需要的状态类应该有以下功能:

  • 每次进入状态的时候执行一些动作,退出时执行一些动作
  • 我们希望让状态在发生改变时能够感知到原因——也就是发生的事件
  • 我们希望每一个状态都有一个自己的唯一编号
  • 我们希望每个状态能够检查是否可以发生指定的事件
  • 状态能够有一个公开给使用者的字典用以保存信息

据此我们设计如下的状态类:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class State:
"""状态类"""
# 静态变量 - 保存已经生成的状态码
__codes: Set[int] = set()

def __repr__(self) -> str:
"""返回 repr 信息"""
return "<Leaf FSM-State code " + str(self.__code) + ">"

def __hash__(self) -> int:
"""使得状态可哈希 - 返回当前状态 hash(code)"""
return hash(self.__code)

def __init__(self, code: int, description: str):
"""
状态类构造函数:
code: 状态对应状态码 - 整型变量
description: 当前状态的描述
extra: 当前状态的额外描述信息
"""
self.__code = code
self.__description = description
self.__extra = dict()
if self.__code in self.__codes:
raise StateExistError("状态码已经存在 - " + int(code))

self.__codes.add(code)
self.__accepted_events: List[Event] = list()

def add(self, event: Event) -> NoReturn:
"""添加当前状态接受的动作"""
self.__accepted_events.append(event)

@property
def code(self) -> int:
"""返回状态码"""
return self.__code

@property
def description(self) -> str:
"""返回状态说明"""
return self.__description

@property
def extra(self) -> dict:
"""返回当前状态的额外信息"""
return self.__extra

@extra.setter
def extra_setter(self, extra: dict) -> NoReturn:
"""设置当前状态的额外信息"""
self.__extra = extra

def accept(self, event: Event) -> bool:
"""检查当前的状态是否接受指定的事件"""
event_class = getattr(event, "__class__", None)
return True if event_class in self.__accepted_events else False

def enter(self, reason: Optional[Event] = None) -> NoReturn:
"""
进入状态时执行的函数 - 接口
传入由什么事件导致进入该状态
"""
pass

def exit(self, reason: Optional[Event] = None) -> NoReturn:
"""
退出当前状态时执行的函数 - 接口
传入因为什么事件导致退出该状态
"""
pass

这里的私有静态变量__codes表示的是一个状态码集合,用以检查新状态的状态码是否冲突;而accept方法用以检查当前状态是否支持发生制定事件;extra字典用于保存额外信息。

事件类

我们需要的事件类应该具有如下功能:

  • 每一个事件都有自己的描述,作为一个静态变量
  • 事件类在实例化时可以生成一个事件的uuid,并且保存发生时间

根据以上的需求,我们设计这样的事件类:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class Event:
"""事件类"""
description: str = str()

def action(self, *args, **kwargs) -> Any:
"""接口函数 - 用于指示事件动作"""
pass

def __repr__(self) -> str:
"""返回 repr 信息"""
return "<Leaf FSM-Event>"

def __hash__(self) -> int:
"""返回当前事件的哈希值 - 函数的hash"""
return hash(self.__opcode)

def __init__(self):
"""
事件类构造函数:
opcode: 当前时间的事件码 - 随机
time: 事件发生的事件
result: 在动作发生之后用于保存动作函数的相关值
"""
self.__time: int = timetools.now()
self.__opcode: str = enctools.random(16)
self.__extra: dict = dict()

def append(self, key, value):
"""给当前的事件添加额外信息描述"""
self.__extra[key] = value

@property
def extra(self) -> dict:
"""返回事件的额外信息"""
return self.__extra

@property
def time(self) -> int:
"""返回事件发生时间"""
return self.__time

@property
def opcode(self):
"""返回事件的操作码"""
return self.__opcode

事件管理器类

状态管理器则是以上两者的结合,我们可以写出他的工作流:

  1. 检查状态转移表,根据事件和状态确定目标状态
  2. 首先检查当前的事件是否被当前状态接受 - accept 方法
  3. 如果接受的话执行当前状态的 exit 函数来退出当前状态
  4. 之后执行事件的转移函数
  5. 下来进入目标状态 - enter 方法

我们在调用状态的 enter/exit 方法时可以将事件传入,作为原因使得状态感知;同时我们可以在类内部维护一个记录器,记录历史发生的所有事件,这在订单状态管理中是非常有用的。

用一张流程图说明会更清晰一些:

下面开始编码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class Machine:
"""状态管理器"""

def __repr__(self) -> str:
"""返回 repr 信息"""
return "<Leaf FSM-Machine '" + self.__name + "'>"

def __init__(self, name: str):
"""
状态管理器:
name: 当前状态机的名称
current: 当前状态
recorder: 事件记录器
transitions: 变化状态表
{
(OriginState, Action): DestState,
...
}
"""
self.__name: str = name
self.__current: State = None
self.__recorder: List[Event] = list()
self.__transitions: Dict[Tuple[State, Event], State] = dict()

@property
def current(self) -> State:
"""返回当前状态"""
return self.__current

@property
def name(self) -> str:
"""返回状态机名称"""
return self.__name

@property
def events(self) -> List[Event]:
"""返回所有发生过的事件"""
return self.__recorder

def handle(self, event: Event, *args, **kwargs) -> NoReturn:
"""
对当前状态改变转移, 当发生了一个指定的事件时, 执行顺序如下:
0. 判断这个事件是否被当前状态接受
1. 如果接受, 执行当前状态的 exit 函数
2. 执行指定事件的 action 函数
3. 执行转移目标状态的 enter 函数
4. 记录发生的事件信息
*args, **kwargs 参数将被传递给当前事件的执行函数
"""
if not self.__current.accept(event):
raise EventNotAccept("当前状态不允许事件该事件发生")

try:
event_class = getattr(event, "__class__", None)
destination: State = self.__transitions[(self.__current, event_class)]
except KeyError as _error:
raise DestinationNotExist("找不到目标状态: " + str(_error.args))

self.__current.exit(event)
event.action(*args, **kwargs)
destination.enter(event)
self.__current = destination
self.__recorder.append(event)

def add(self, origin: State, event: Event, destination: State) -> NoReturn:
"""给状态转移表中添加一条记录"""
self.__transitions[(origin, event)] = destination

def start(self, event: Event) -> NoReturn:
"""启动状态机运行初始进入函数"""
self.__current = event
self.__current.enter()

def stop(self) -> NoReturn:
"""当确定状态机到达最终状态时可以执行该函数用于退出最终状态"""
self.__current.exit()
self.__current = None

下面一篇文章,我们将使用以上的状态机来管理系统订单状态。

Author: 桂小方

Permalink: https://init.blog/1833/

文章许可协议:

如果你觉得文章对你有帮助,可以 支持我

Comments