Python实现状态机有多种方式，从简单的字典/类实现到使用功能强大的第三方库。下面通过具体示例对比不同实现方案，并分析最佳实践。 ## 一、Python状态机实现方案对比 | 实现方式 | 核心思想 | 优点 | 缺点 | 适用场景 | |---------|---------|------|------|---------| | **基于字典/表的简单实现** | 使用字典存储状态转换规则 | 简单直观，无需依赖 | 状态逻辑分散，不易维护复杂逻辑 | 简单流程，状态<10个 | | **基于类的状态模式** | 每个状态封装为独立类 | 符合OOP原则，易扩展 | 类数量多，代码量较大 | 中型项目，状态行为差异大 | | **使用transitions库** | 声明式状态机框架 | 功能强大，支持分层、异步等 | 需要学习库API | 生产环境，复杂业务逻辑 | | **基于生成器的协程状态机** | 使用yield实现状态流转 | 简洁，天然支持异步 | 调试复杂，状态持久化困难 | I/O密集型异步任务 | ## 二、具体实现示例 ### 1. 基于字典的简单状态机（订单系统示例） ```python class SimpleOrderStateMachine: """基于字典的简单订单状态机实现""" def __init__(self): # 定义状态 self.states = { 'pending': '待支付', 'paid': '已支付', 'shipped': '已发货', 'delivered': '已签收', 'cancelled': '已取消' } # 定义状态转换规则：当前状态 -> {事件: 目标状态} self.transitions = { 'pending': { 'pay': 'paid', 'cancel': 'cancelled' }, 'paid': { 'ship': 'shipped', 'refund': 'pending' }, 'shipped': { 'deliver': 'delivered' }, 'delivered': { 'return': 'pending' }, 'cancelled': {} # 终止状态，无转换 } self.current_state = 'pending' self.history = [] # 状态历史记录 def trigger(self, event, **kwargs): """触发状态转换""" if event not in self.transitions[self.current_state]: raise ValueError(f"状态 {self.current_state} 不支持事件 {event}") old_state = self.current_state self.current_state = self.transitions[old_state][event] # 记录状态转换 transition = { 'from': old_state, 'to': self.current_state, 'event': event, 'timestamp': kwargs.get('timestamp'), 'data': kwargs } self.history.append(transition) # 执行对应的动作 action_name = f"on_{event}" if hasattr(self, action_name): getattr(self, action_name)(**kwargs) print(f"状态转换: {old_state} -> {self.current_state} (事件: {event})") return True def on_pay(self, **kwargs): """支付事件处理""" amount = kwargs.get('amount', 0) print(f"处理支付: 金额 {amount}元") def on_ship(self, **kwargs): """发货事件处理""" tracking_number = kwargs.get('tracking_number', '') print(f"处理发货: 物流单号 {tracking_number}") def get_possible_events(self): """获取当前状态允许的事件""" return list(self.transitions[self.current_state].keys()) # 使用示例 if __name__ == "__main__": order = SimpleOrderStateMachine() print(f"初始状态: {order.current_state}") print(f"允许的事件: {order.get_possible_events()}") order.trigger('pay', amount=100.0, user_id=123) order.trigger('ship', tracking_number='SF123456789') order.trigger('deliver') print(f"\n状态历史:") for trans in order.history: print(f" {trans['from']} -> {trans['to']} via {trans['event']}") ``` ### 2. 基于状态模式的高级实现 ```python from abc import ABC, abstractmethod from datetime import datetime class OrderState(ABC): """订单状态基类""" @abstractmethod def pay(self, order, **kwargs): pass @abstractmethod def cancel(self, order, **kwargs): pass @abstractmethod def ship(self, order, **kwargs): pass @abstractmethod def deliver(self, order, **kwargs): pass @abstractmethod def get_name(self): pass class PendingState(OrderState): """待支付状态""" def pay(self, order, **kwargs): print("执行支付逻辑...") # 验证支付信息 # 调用支付接口 order.change_state(PaidState()) return True def cancel(self, order, **kwargs): print("订单取消...") order.change_state(CancelledState()) return True def ship(self, order, **kwargs): raise ValueError("待支付状态不能发货") def deliver(self, order, **kwargs): raise ValueError("待支付状态不能签收") def get_name(self): return "pending" class PaidState(OrderState): """已支付状态""" def pay(self, order, **kwargs): raise ValueError("已支付状态不能再次支付") def cancel(self, order, **kwargs): # 检查是否允许取消（如未发货可取消） if kwargs.get('force', False): print("强制取消订单...") order.change_state(CancelledState()) return True return False def ship(self, order, **kwargs): print("执行发货逻辑...") order.change_state(ShippedState()) return True def deliver(self, order, **kwargs): raise ValueError("已支付状态不能直接签收") def get_name(self): return "paid" class ShippedState(OrderState): """已发货状态""" # 类似实现其他状态类... class Order: """订单上下文类""" def __init__(self, order_id): self.order_id = order_id self.state = PendingState() # 初始状态 self.state_history = [] self.created_at = datetime.now() def change_state(self, new_state): """状态转换""" old_state_name = self.state.get_name() new_state_name = new_state.get_name() # 记录状态历史 self.state_history.append({ 'from': old_state_name, 'to': new_state_name, 'time': datetime.now() }) self.state = new_state print(f"订单 {self.order_id}: {old_state_name} -> {new_state_name}") def pay(self, **kwargs): return self.state.pay(self, **kwargs) def cancel(self, **kwargs): return self.state.cancel(self, **kwargs) def ship(self, **kwargs): return self.state.ship(self, **kwargs) def get_current_state(self): return self.state.get_name() # 使用示例 if __name__ == "__main__": order = Order("ORD20231231001") print(f"订单状态: {order.get_current_state()}") try: order.pay(amount=199.0, payment_method="alipay") order.ship(tracking_number="SF987654321") # order.pay() # 这里会抛出异常 except ValueError as e: print(f"操作失败: {e}") ``` ### 3. 使用transitions库的生产级实现 ```python from transitions import Machine from transitions.extensions import HierarchicalMachine # 分层状态机 import json class OrderSystem: """使用transitions库的完整订单管理系统""" states = [ 'draft', # 草稿 {'name': 'pending', 'on_enter': 'on_pending_enter'}, # 待支付 'paid', # 已支付 {'name': 'processing', 'children': ['packing', 'shipping']}, # 处理中（父状态） 'shipped', # 已发货 'delivered', # 已签收 'cancelled', # 已取消 'refunded' # 已退款 ] transitions = [ # 从 草稿 到 待支付 {'trigger': 'submit', 'source': 'draft', 'dest': 'pending', 'before': 'validate_order'}, # 支付相关转换 {'trigger': 'pay', 'source': 'pending', 'dest': 'paid', 'conditions': 'check_payment', 'after': 'send_payment_success'}, {'trigger': 'pay_failed', 'source': 'pending', 'dest': 'draft'}, # 发货流程 {'trigger': 'process', 'source': 'paid', 'dest': 'processing'}, {'trigger': 'pack', 'source': 'processing_packing', 'dest': 'processing_shipping'}, {'trigger': 'ship', 'source': 'processing_shipping', 'dest': 'shipped', 'after': 'generate_shipping_label'}, # 签收和退货 {'trigger': 'deliver', 'source': 'shipped', 'dest': 'delivered'}, {'trigger': 'request_return', 'source': 'delivered', 'dest': 'processing', 'conditions': 'within_return_period'}, # 取消和退款 {'trigger': 'cancel', 'source': ['draft', 'pending'], 'dest': 'cancelled'}, {'trigger': 'refund', 'source': ['paid', 'delivered'], 'dest': 'refunded', 'conditions': 'refund_approved'} ] def __init__(self, order_id): self.order_id = order_id self.amount = 0 self.payment_info = None self.shipping_info = None # 创建分层状态机 self.machine = HierarchicalMachine( model=self, states=OrderSystem.states, transitions=OrderSystem.transitions, initial='draft', ignore_invalid_triggers=True # 忽略无效触发 ) def validate_order(self): """订单提交前的验证""" print(f"验证订单 {self.order_id}...") return True def check_payment(self): """支付条件检查""" if self.amount <= 0: print("订单金额必须大于0") return False return True def send_payment_success(self): """支付成功后的操作""" print(f"订单 {self.order_id} 支付成功，金额: {self.amount}") # 发送通知、更新库存等 def generate_shipping_label(self): """生成物流标签""" print(f"为订单 {self.order_id} 生成物流标签") def within_return_period(self): """检查是否在退货期内""" return True # 简化实现 def refund_approved(self): """退款审批检查""" return True # 简化实现 def on_pending_enter(self): """进入待支付状态时的操作""" print(f"订单 {self.order_id} 进入待支付状态") def to_dict(self): """序列化状态机状态""" return { 'order_id': self.order_id, 'state': self.state, 'amount': self.amount, 'is_paid': self.is_paid() # transitions自动生成的方法 } # 使用示例 if __name__ == "__main__": # 创建订单 order = OrderSystem("ORD20240101001") order.amount = 100.0 print(f"初始状态: {order.state}") # 状态转换 order.submit() print(f"提交后状态: {order.state}") order.pay() print(f"支付后状态: {order.state}") order.process() print(f"处理后状态: {order.state}") # 查看所有可能的转换 print(f"\n当前状态允许的转换: {order.machine.get_triggers(order.state)}") # 可视化状态机（需要安装graphviz） # order.machine.get_graph().draw('order_state_diagram.png', prog='dot') ``` ## 三、Python状态机最佳实践 ### 1. 状态机设计原则 ```python from enum import Enum from typing import Dict, List, Optional, Callable from dataclasses import dataclass @dataclass class Transition: """状态转换定义""" event: str source: str target: str condition: Optional[Callable] = None action: Optional[Callable] = None class StateMachineDesign: """ 状态机设计最佳实践 1. 明确状态边界 2. 处理异常转换 3. 支持状态持久化 4. 提供监控和调试接口 """ def __init__(self): # 使用枚举定义状态，避免魔法字符串 class States(Enum): INIT = "init" PROCESSING = "processing" SUCCESS = "success" FAILED = "failed" self.States = States self.current_state = States.INIT self.transition_history: List[Dict] = [] # 定义状态转换表 self.transition_table: Dict[Enum, List[Transition]] = { States.INIT: [ Transition("start", States.INIT, States.PROCESSING, condition=self._can_start, action=self._on_start) ], States.PROCESSING: [ Transition("complete", States.PROCESSING, States.SUCCESS, action=self._on_complete), Transition("fail", States.PROCESSING, States.FAILED, action=self._on_fail) ] } def _can_start(self) -> bool: """条件检查示例""" return True def _on_start(self, **kwargs): """转换动作示例""" print("任务开始执行...") def trigger(self, event: str, **kwargs) -> bool: """安全的触发状态转换""" current_state = self.current_state if current_state not in self.transition_table: return False # 查找匹配的转换 for transition in self.transition_table[current_state]: if transition.event == event: # 检查条件 if transition.condition and not transition.condition(**kwargs): print(f"条件检查失败: {event}") return False # 执行动作 old_state = self.current_state self.current_state = transition.target if transition.action: transition.action(**kwargs) # 记录历史 self.transition_history.append({ 'timestamp': datetime.now(), 'from': old_state.value, 'to': self.current_state.value, 'event': event, 'data': kwargs }) print(f"状态转换成功: {old_state.value} -> {self.current_state.value}") return True print(f"未找到匹配的转换: {current_state.value} + {event}") return False ``` ### 2. 异步状态机实现 ```python import asyncio from transitions.extensions.asyncio import AsyncMachine class AsyncOrderProcessor: """异步状态机示例""" states = ['idle', 'processing', 'completed', 'failed'] def __init__(self): self.data = None # 创建异步状态机 self.machine = AsyncMachine( model=self, states=self.states, initial='idle', transitions=[ ['process', 'idle', 'processing'], ['success', 'processing', 'completed'], ['error', 'processing', 'failed'], ['reset', ['completed', 'failed'], 'idle'] ] ) async def on_enter_processing(self): """进入处理状态的异步操作""" print("开始异步处理...") await asyncio.sleep(1) # 模拟异步操作 self.data = await self.fetch_data() async def fetch_data(self): """模拟异步数据获取""" await asyncio.sleep(0.5) return {"result": "success"} async def main(): processor = AsyncOrderProcessor() print(f"初始状态: {processor.state}") # 异步触发状态转换 await processor.process() print(f"处理中状态: {processor.state}") if processor.data: await processor.success() else: await processor.error() print(f"最终状态: {processor.state}") # 运行: asyncio.run(main()) ``` ## 四、状态机在真实场景中的应用 ### 1. 工作流引擎实现 ```python class WorkflowEngine: """基于状态机的工作流引擎""" def __init__(self, workflow_definition: Dict): """ workflow_definition 示例: { "states": ["draft", "review", "approved", "rejected"], "transitions": [ { "name": "submit", "from": "draft", "to": "review", "conditions": ["is_valid"], "actions": ["notify_reviewer"] } ], "initial_state": "draft" } """ self.definition = workflow_definition self.current_state = workflow_definition['initial_state'] self.context = {} # 工作流上下文数据 def can_transition(self, transition_name: str) -> bool: """检查是否可以执行转换""" # 实现条件检查逻辑 pass def execute_transition(self, transition_name: str, **kwargs): """执行状态转换""" # 实现转换逻辑，包括条件检查、动作执行 pass ``` ### 2. 游戏角色状态机 ```python class CharacterStateMachine: """游戏角色状态机""" def __init__(self, character): self.character = character self.st