什么是状态机?
关键属性: 状态和转换
状态: 系统当前状态
转换:一种状态到另外一种状态的变化。
转换由触发事件或是条件启动。
状态机-状态图
状态机使用场景:
自动售货机 电梯 交通灯 组合锁 停车计时器
使用state_machine 模块创建状态机第一步使用@acts_as_state_machine装饰器
@acts_as_state_machine
class Process:
initial 属性值设置为 True
created = State(initial=True) waiting = State() running = State() terminated = State() blocked = State() swapped_out_waiting = State() swapped_out_blocked = State()
定义转换。在 state_machine 模块中,转换是 Event 类的一个实例。我们使用
参数 from_states 和 to_state 定义可能的转换。
wait = Event(from_states=(created, running, blocked, swapped_out_waiting), to_state=waiting) run = Event(from_states=waiting, to_state=running) terminate = Event(from_states=running, to_state=terminated) block = Event(from_states=(running, swapped_out_blocked), to_state=blocked) swap_wait = Event(from_states=waiting, to_state=swapped_out_waiting) swap_block = Event(from_states=blocked, to_state=swapped_out_blocked)
from_states 可以是单个状态,也可以是一组状态(元组)。
state_machine 模块为我们
提供了 @before 和 @after 装饰器,二者可以分别用于在转换发生之前或之后执行操作。你可以
想象在系统中更新一些对象,或者向某人发送电子邮件或通知。在本例中,操作仅限于打印关于
进程状态更改的信息。
transition() 函数,它接受三个参数:
process , Process 的一个实例;
event , Event 的一个实例( wait 、 run 、 terminate 等);
event_name ,事件的名称。
执行事件时出错,则输出事件的名称。
下面是 transition() 函数的代码:
def transition(process, event, event_name): try: event() except InvalidStateTransition as err: print(f'Error: transition of {process.name} from {process.current_state} to {event_name} failed')
state_info() 函数显示进程当前(激活)状态的一些基本信息。
def state_info(process): print(f'state of {process.name}: {process.current_state}')
在 main() 函数的开头,我们定义了一些字符串常量,它们被作为 event_name 传递。
def main(): RUNNING = 'running' WAITING = 'waiting' BLOCKED = 'blocked' TERMINATED = 'terminated'
创建两个 Process 实例并展示它们的初始状态信息。
p1, p2 = Process('process1'), Process('process2') [state_info(p) for p in (p1, p2)]
允许的转换应该与
状态图相关。例如,应该可以从一个运行状态切换到一个阻塞状态,但是不应该从一个阻塞状态
切换到一个运行状态。
from state_machine import (State,Event,acts_as_state_machine,after,before,InvalidStateTransition) @acts_as_state_machine class Process: created=State(initial=True) # 创建状态 waiting=State() #等待状态 running=State()# 运行状态 terminated=State()# 停止状态 blocked=State() # 阻塞 swapper_out_waiting=State()# swapper_out_blocked=State() # 等待状态 转入的状态from_states ,目标状态: to_state wait=Event(from_states=(created,running,blocked,swapper_out_waiting),to_state=waiting) run=Event(from_states=waiting,to_state=running) terminate=Event(from_states=running,to_state=terminated) block=Event(from_states=(running,swapper_out_blocked),to_state=blocked) swap_wait=Event(from_states=waiting,to_state=swapper_out_waiting) swap_block=Event(from_states=blocked,to_state=swapper_out_blocked) def __init__(self,name): self.name=name @after('wait') def wait_info(self): print(f'{self.name} entered waiting mode') @after('run') def run_info(self): print(f'{self.name} is running') @before('ternimate') def terminate_info(self): print(f"{self.name} terminated") @after('block') def block_info(self): print(f'{self.name} is blocked') @after('swap_wait') def swap_wait_info(self): print(f'{self.name} is swapped out and waiting') @after('swap_block') def swap_block_info(self): print(f'{self.name} is swapped out and blocked') @after('block') def block_info(self): print(f'{self.name} is blocked') @after('swap_wait') def swap_wait_info(self): print(f'{self.name} is swapped out and waiting') @after('swap_block') def swap_block_info(self): print(f'{self.name} is swapped out and blocked') def transition(process,event,event_name): try: event() except InvalidStateTransition as err: print(f"Error: transaction of {process.name} from {process.current_state} to {event_name} failed") # 显示信息 def state_info(process): print(f'state of {process.name}:{process.current_state}') def main(): RUNNING='running' WAITING='waiting' BLOCKED='blocked' TERMINATED='terminated' p1,p2=Process('process1'),Process('process2') [state_info(p) for p in (p1,p2)] print("-------1----------") transition(p1,p1.wait,WAITING) transition(p2,p2.terminate,TERMINATED) [state_info(p) for p in (p1,p2)] print("------2----------") transition(p1,p1.run,RUNNING) transition(p2,p2.wait,WAITING) [state_info(p) for p in (p1, p2)] print("------3----------") transition(p2, p2.run, RUNNING) [state_info(p) for p in (p1, p2)] print("------4----------") [transition(p,p.block,BLOCKED) for p in (p1,p2)] [state_info(p) for p in (p1, p2)] print("------5----------") [transition(p, p.terminate, TERMINATED) for p in (p1, p2)] if __name__=='__main__': main() state of process1:created state of process2:created -------1---------- process1 entered waiting mode Error: transaction of process2 from created to terminated failed state of process1:waiting state of process2:created ------2---------- process1 is running process2 entered waiting mode state of process1:running state of process2:waiting ------3---------- process2 is running state of process1:running state of process2:running ------4---------- process1 is blocked process1 is blocked process2 is blocked process2 is blocked state of process1:blocked state of process2:blocked ------5---------- Error: transaction of process1 from blocked to terminated failed Error: transaction of process2 from blocked to terminated failed