3、Python实现状态机：面向对象的状态模式、使用字典驱动状态机、第三方库transitions介绍
好，咱们进入实战环节。
前面两章讲了理论，讲了协议拆解。但说真的，纸上谈兵没意思。我当年刚入行的时候，看了一堆状态机论文，结果写出来的代码又臭又长。后来才明白——状态机的实现，其实就三种主流玩法。
今天我把它们全盘托出。
3.1 面向对象的状态模式
这是最正统的玩法。设计模式里的State Pattern，说白了就是把每个状态封装成一个类。
为什么要这么做？
你想想看，在交易所协议解析里，每个状态的行为差异很大。比如「等待登录」状态和「已登录」状态，对同一个消息的处理逻辑完全不同。用if-else硬写？代码会变成一坨浆糊。
核心思想：状态本身也是对象。每个状态对象负责自己的行为逻辑。
我在项目中遇到过最典型的场景——FIX协议的会话层。登录、心跳、重连、登出，每个状态的处理逻辑都不一样。用状态模式，代码结构清晰得像教科书。
from abc import ABC, abstractmethod
class SessionState(ABC):
"""会话状态基类"""
@abstractmethod
def on_message(self, context, msg):
pass
@abstractmethod
def on_timeout(self, context):
pass
class LoggingInState(SessionState):
"""登录中状态"""
def on_message(self, context, msg):
if msg.type == 'Logon':
print("登录成功，切换到已登录状态")
context.set_state(LoggedInState())
else:
print("登录失败，保持登录中状态")
def on_timeout(self, context):
print("登录超时，切换到断开状态")
context.set_state(DisconnectedState())
class LoggedInState(SessionState):
"""已登录状态"""
def on_message(self, context, msg):
if msg.type == 'Logout':
print("收到登出请求，切换到登出中状态")
context.set_state(LoggingOutState())
elif msg.type == 'Heartbeat':
print("处理心跳消息")
else:
print(f"处理业务消息: {msg.type}")
def on_timeout(self, context):
print("心跳超时，发送心跳请求")
context.send_heartbeat()
class SessionContext:
"""会话上下文，持有当前状态"""
def __init__(self):
self._state = LoggingInState()
def set_state(self, state):
self._state = state
def handle_message(self, msg):
self._state.on_message(self, msg)
def handle_timeout(self):
self._state.on_timeout(self)我的经验：状态模式适合状态数量固定、状态间转换逻辑复杂的场景。但别滥用——如果只有两三个状态，用这个模式反而显得笨重。
3.2 使用字典驱动状态机
这是我最喜欢的方式。轻量、高效、一目了然。
说白了，就是用字典把「当前状态 + 事件 -> 下一个状态 + 动作」这个映射关系存起来。Python的字典天生就是干这个的。
我记得有一次做高频交易系统的协议解析，要求延迟在微秒级。状态模式那种对象创建的开销，根本扛不住。字典驱动的方式，直接查表，性能好得多。
class DictDrivenStateMachine:
"""字典驱动的状态机"""
def __init__(self):
self.state = 'INIT'
self.transitions = {
('INIT', 'CONNECT'): ('CONNECTING', self._on_connect),
('CONNECTING', 'SUCCESS'): ('CONNECTED', self._on_connected),
('CONNECTING', 'FAIL'): ('INIT', self._on_connect_fail),
('CONNECTED', 'LOGIN'): ('LOGGING_IN', self._on_login),
('LOGGING_IN', 'LOGIN_OK'): ('READY', self._on_login_ok),
('LOGGING_IN', 'LOGIN_FAIL'): ('CONNECTED', self._on_login_fail),
('READY', 'HEARTBEAT'): ('READY', self._on_heartbeat),
('READY', 'TIMEOUT'): ('RECONNECTING', self._on_timeout),
('RECONNECTING', 'RECONNECT_OK'): ('CONNECTED', self._on_reconnect_ok),
('RECONNECTING', 'RECONNECT_FAIL'): ('DISCONNECTED', self._on_disconnect),
}
def _on_connect(self, event_data):
print(f"发起连接: {event_data}")
def _on_connected(self, event_data):
print(f"连接成功: {event_data}")
def _on_login(self, event_data):
print(f"发送登录请求: {event_data}")
def _on_login_ok(self, event_data):
print(f"登录成功: {event_data}")
def _on_heartbeat(self, event_data):
print(f"处理心跳: {event_data}")
def _on_timeout(self, event_data):
print(f"超时处理: {event_data}")
def _on_reconnect_ok(self, event_data):
print(f"重连成功: {event_data}")
def _on_disconnect(self, event_data):
print(f"断开连接: {event_data}")
def _on_connect_fail(self, event_data):
print(f"连接失败: {event_data}")
def _on_login_fail(self, event_data):
print(f"登录失败: {event_data}")
def trigger(self, event, event_data=None):
key = (self.state, event)
if key in self.transitions:
next_state, action = self.transitions[key]
action(event_data)
self.state = next_state
print(f"状态变更: {key[0]} -> {next_state}")
else:
print(f"非法转换: 当前状态={self.state}, 事件={event}")避坑指南：我曾经在字典驱动状态机里忘了处理所有可能的转换组合，结果线上出现了「状态卡死」的情况。后来我加了一个validate_transitions()方法，在初始化时自动检查所有状态是否都有对应的处理逻辑。
3.3 第三方库transitions介绍
如果你不想自己造轮子，transitions这个库值得一试。
我第一次用这个库的时候，说实话有点惊艳。几行代码就能搭出一个完整的状态机。而且它支持分层状态、并行状态这些高级特性。
from transitions import Machine
import random
class TradingSession:
"""交易会话状态机"""
states = ['disconnected', 'connecting', 'connected', 'logging_in',
'ready', 'reconnecting', 'disconnecting']
def __init__(self):
self.machine = Machine(
model=self,
states=TradingSession.states,
initial='disconnected',
auto_transitions=False
)
# 定义转换
self.machine.add_transition('connect', 'disconnected', 'connecting',
after='on_connect')
self.machine.add_transition('connect_success', 'connecting', 'connected',
after='on_connected')
self.machine.add_transition('connect_fail', 'connecting', 'disconnected',
after='on_connect_fail')
self.machine.add_transition('login', 'connected', 'logging_in',
after='on_login')
self.machine.add_transition('login_ok', 'logging_in', 'ready',
after='on_login_ok')
self.machine.add_transition('login_fail', 'logging_in', 'connected',
after='on_login_fail')
self.machine.add_transition('heartbeat', 'ready', 'ready',
after='on_heartbeat')
self.machine.add_transition('timeout', 'ready', 'reconnecting',
after='on_timeout')
self.machine.add_transition('reconnect_ok', 'reconnecting', 'connected',
after='on_reconnect_ok')
self.machine.add_transition('reconnect_fail', 'reconnecting', 'disconnected',
after='on_disconnect')
self.machine.add_transition('disconnect', '*', 'disconnecting',
after='on_disconnect')
self.machine.add_transition('disconnected', 'disconnecting', 'disconnected',
after='on_disconnected')
def on_connect(self):
print("发起连接...")
def on_connected(self):
print("连接建立成功")
def on_connect_fail(self):
print("连接失败")
def on_login(self):
print("发送登录请求...")
def on_login_ok(self):
print("登录成功，准备就绪")
def on_heartbeat(self):
print("处理心跳消息")
def on_timeout(self):
print("超时，准备重连")
def on_reconnect_ok(self):
print("重连成功")
def on_disconnect(self):
print("断开连接...")
def on_disconnected(self):
print("已断开")
# 使用示例
session = TradingSession()
print(f"初始状态: {session.state}")
session.connect()
print(f"当前状态: {session.state}")
# 模拟连接成功
session.connect_success()
print(f"当前状态: {session.state}")
# 模拟登录
session.login()
session.login_ok()
print(f"当前状态: {session.state}")
# 模拟心跳
session.heartbeat()
print(f"当前状态: {session.state}")transitions的优势：
- 支持条件转换（只有满足条件才触发）
- 支持转换前后的回调函数
- 支持状态进入/离开时的钩子函数
- 内置状态图可视化功能
- 支持分层状态机（状态里嵌套状态机）
3.4 三种方式的对比
| 特性 | 状态模式 | 字典驱动 | transitions |
|---|---|---|---|
| 代码可读性 | 高 | 中 | 高 |
| 性能 | 中（有对象开销） | 高（直接查表） | 中 |
| 灵活性 | 中 | 高 | 高 |
| 学习成本 | 中 | 低 | 低 |
| 适合场景 | 状态少、逻辑复杂 | 性能敏感、状态多 | 快速原型、复杂状态机 |
3.5 我的选择建议
嗯，这里要注意。没有银弹。
我个人习惯是这样：
- 做原型验证：直接用transitions，省时间
- 做高频交易系统：用字典驱动，性能第一
- 做复杂的业务逻辑：用状态模式，方便扩展和维护
我曾经在一个项目里，三种方式混着用。核心链路用字典驱动，外围逻辑用transitions，复杂的业务状态用状态模式。效果还不错。
最后说一句：状态机的实现方式不重要，重要的是你理解了「状态」和「转换」这两个核心概念。工具只是工具，脑子里的模型才是关键。
小提示：不管你选哪种方式，先把状态转换图画出来。画图的过程，就是理清逻辑的过程。我每次做新协议解析，第一件事就是拿白板画状态图。