Python 状态机在业务流程中的应用

1次阅读

简单流转用 dict,复杂逻辑(副作用、历史、回滚)必须用 class;状态变更需原子化封装、与存储解耦;历史记录须独立建表并索引;异步任务要区分“决策完成”与“动作完成”,通过幂等和最终一致性保障可靠。

Python 状态机在业务流程中的应用

状态机该用 class 还是 dict 实现

业务流程里状态跳转不复杂时,dict 足够——比如订单从 "created""paid""shipped",用一个映射表就能说清合法转移。但一旦涉及动作钩子(比如付款后要发消息、发货前要校验库存)、状态进入/退出逻辑,或者需要记录历史、支持回滚,class 就绕不开。

常见错误是硬塞所有逻辑进 dict 值里,比如把函数引用塞进字典再手动调用,结果调试时找不到执行路径、异常断层。也有人一上来就写十几层继承的状态机框架,结果连“取消订单”这种单步操作都要配 5 个配置项。

  • 简单流转(≤5 个状态、无副作用):用 {state: {Event: next_state}} 结构,配合一个 transition(event) 方法即可
  • 需执行副作用(如调用 API、改数据库):每个状态建一个方法,用 getattr(self, f"on_enter_{state}")@state_transition 装饰器统一调度
  • 别在状态跳转函数里直接写 DB commit;把“状态变更”和“持久化”拆开,否则测试时 mock 困难、事务边界模糊

如何让状态变更真正原子化

线上最常踩的坑不是逻辑写错,而是状态设了但没存住,或存了但其他字段没同步更新,导致数据库里状态是 "shipped",实际物流单号还是空。python 层面的 self.state = "shipped" 不等于落地。

关键不在状态机本身,而在它和数据存储的耦合方式。ORM 如 SQLAlchemy 可以用 before_update 事件拦截,但更稳的做法是把状态变更封装成一个函数,内部做校验 + 更新 + 日志,然后整个包进事务。

立即学习Python免费学习笔记(深入)”;

  • 永远通过一个入口方法触发变更,例如 order.transition("ship", tracking_no="SF123"),而不是暴露 order.state = ...
  • 该方法内第一件事是查当前状态是否允许这次事件(防止重复发货),第二件事才是更新字段和关联数据
  • 如果用 django,别依赖 save() 的信号机制做状态后续动作;信号可能被禁用、顺序难控,改用显式调用 on_shipped()

状态机要不要存历史记录

要不要存,取决于你能否承受“不知道谁、什么时候、为什么把订单从待支付改成已取消”。客服查问题、审计合规、甚至排查自己写的 bug,90% 都靠这条链路。

很多人以为加个 StateHistory 模型就行,结果每次状态变都 insert 一条,没加索引,半年后查一个订单的历史要 8 秒。也有人把历史全塞进 json 字段,看着省事,结果没法按时间范围、事件类型筛选。

  • 必须单独建表,至少含 order_idfrom_stateto_stateeventcreated_atoperator_id
  • 在状态变更主方法末尾统一写入,别分散在各处;避免漏记、重复记
  • 如果业务允许,给历史表加复合索引:(order_id, created_at)(event, created_at)

异步任务中状态机容易断在哪

用户下单后发短信、通知仓库、更新搜索索引——这些常丢进 Celery。问题在于:状态机认为“已支付”,但发短信的任务失败重试了三次才成功,期间客服系统看到的仍是旧状态,而仓库已经收到了出库指令。

根本矛盾是状态推进和异步动作的生命周期不一致。不能指望 Celery 任务一定成功,也不能让主流程等它全部做完。

  • 状态变更只反映“决策完成”,不保证“动作完成”。例如 "paid" 表示支付确认已受理,不是“短信已发完”
  • 对必须成功的异步动作(如扣减库存),改用本地事务+延迟队列(如 django-qschedule)或两阶段提交模式
  • 给异步任务加幂等键,比如 task_id = f"send_sms_{order_id}_{order.version}",避免状态反复变更导致任务重复触发

状态机真正的难点从来不在状态怎么定义,而在于你有没有想清楚:哪部分必须强一致,哪部分可以最终一致,以及当不一致发生时,靠什么手段发现和修复。这和代码写多漂亮关系不大,和日志打不全、监控埋点漏没漏,关系很大。

text=ZqhQzanResources