从if-else到状态机：手写实现与工程实践

1. 从一把 if-else 里逃出来

大概两年前，我接手过一个内部管理系统的订单模块。代码本身不算复杂，但状态相关的逻辑全部揉在一堆布尔标志位和嵌套 if-else 里。那种代码你要说完全跑不了，也不至于，但每加一个新需求，我就要小心翼翼地看一遍“现在是哪些标志位组合，下一步该走哪条分支”。有一次线上订单卡在“已支付但未发货”，查了半天才发现是某个回调事件把状态又翻回了旧值。

这段经历基本就是我开始认真用状态机的原因。State Machines 这个词在教科书里出现得次数不少，但真正让我觉得“这不是理论，是救命工具”的，是把它拆成“状态 + 事件 + 转移表”这三个零件以后，代码的可读性和可控性一下子变得完全不一样。

这篇文章适合两类人。一类是维护过那种“标志位泥潭”的开发者，想找一个不引入重量级框架的折中方案；另一类是听过状态机但始终觉得它和实际业务隔了一层的人。我会用订单、任务、连接这类常见模型做例子，给出可以直接拿去改的代码，也会把那些文档里不写的坑一并说清楚。

2. 状态机的三个零件：状态、事件、转移表

2.1 先把术语说人话

状态机听起来吓人，本质就是一句话：系统在任意时刻处于某一个状态，来一个事件，查一下“当前状态 + 事件”对应的转移，然后决定去哪个新状态。

这里最核心的是“当前状态只有一个”。这个约束听起来简单，但正是它把无数种标志位组合压缩成了一条确定的路径。比如“已支付”“已发货”“已完成”这些互斥的概念，你在传统写法里可能用三个布尔变量来表示，结果冒出“既已支付又已发货”的中间态，其实不是设计出来的，是巧合跑出来的。状态机不给你这个机会。

2.2 用订单系统的例子串一遍

拿订单来说，我一般会先定义状态：PENDING（待支付）、PAID（已支付）、SHIPPED（已发货）、COMPLETED（已完成）、CANCELLED（已取消）。再定义事件：PAY_SUCCESS、SHIP、CONFIRM_RECEIPT、CANCEL。然后画一张二维表，行是状态，列是事件，交叉格写转移结果。

这张表的好处是：一眼就能看出哪些转移是允许的，哪些是非法。比如 PENDING + ship 就应该是不允许的，你不可能没付款就发货。传统 if-else 里，这种非法路径往往要靠程序员自觉去堵；在状态机里，它是表格的一个空格，天然暴露在那里。

2.3 转移表怎么设计才不容易漏

设计转移表最容易漏的不是“正常路径”，而是“用户取消”“超时关闭”“对账失败”这类异常路径。我有个习惯：先列出所有状态，再列出所有事件，然后逐个看“每一个状态遇到每一个事件，应该去哪”。可以写一个矩阵或者表格，把不可能出现的格子直接标成 ILLEGAL，而不是留空。

留空的问题在于，代码里一旦遇到没有定义的转移，你通常只能假装没看到；标成 ILLEGAL 之后，你就必须在运行时和日志里明确暴露它。很多线上问题就是这种“没定义但实际发生了”的转移造成的。

3. 不装框架，手写一个最小可用状态机

3.1 为什么我建议先手写

市面上有现成的状态机库，比如 Spring StateMachine、XState，功能很全，但我个人建议第一版先在业务代码里手写一个几十行的转移引擎。原因很简单：库会掩盖概念。当你自己写的时候，你被迫想清楚状态怎么存、事件怎么来、转移时动作放哪里。这些思考比任何框架的 API 都重要。

而且我的经验是，大部分业务场景的状态机不超过十来个状态，手写 30 行代码，比引入一个几百 KB 的库然后处理它的配置语法和维护成本要划算得多。

3.2 核心代码：30 行以内的转移引擎

一个最小状态机，用 Python 写大概是这样的：

from dataclasses import dataclass from typing import Callable, Dict, Tuple, Optional State = str Event = str @dataclass class Transition: target: State action: Optional[Callable] = None class SimpleStateMachine: def __init__(self, initial: State, transitions: Dict[Tuple[State, Event], Transition]): self.state = initial self.transitions = transitions def send(self, event: Event, *args, **kwargs): key = (self.state, event) trans = self.transitions.get(key) if trans is None: raise ValueError(f"Illegal transition: {self.state} + {event}") if trans.action: trans.action(*args, **kwargs) self.state = trans.target

这段代码的核心就三件事：用(state, event)做键查表、有定义就走动作并换状态、没定义就抛异常。没有魔法，没有状态栈，没有复杂继承，但已经能覆盖绝大多数“单层状态机”的业务需求。

3.3 接入业务逻辑后长什么样

假设是订单支付成功回调，状态定义和转移表可以写成：

class OrderState: PENDING = "PENDING" PAID = "PAID" SHIPPED = "SHIPPED" COMPLETED = "COMPLETED" CANCELLED = "CANCELLED" def on_paid(order_id): order = load_order(order_id) order.paid_at = now() save_order(order) def on_ship(order_id, express_no): order = load_order(order_id) order.express_no = express_no save_order(order) transitions = { (OrderState.PENDING, 'pay_success'): Transition(OrderState.PAID, on_paid), (OrderState.PAID, 'ship'): Transition(OrderState.SHIPPED, on_ship), (OrderState.SHIPPED, 'confirm_receipt'): Transition(OrderState.COMPLETED, on_complete), (OrderState.PENDING, 'cancel'): Transition(OrderState.CANCELLED, on_cancel), (OrderState.PAID, 'cancel'): Transition(OrderState.CANCELLED, on_refund), } m = SimpleStateMachine(OrderState.PENDING, transitions) m.send('pay_success', order_id)

这里我把动作放在转移对象里，而不是放在状态里，是因为同一个事件在不同状态下要做的副作用不同。比如 PENDING 状态下取消订单是直接关闭，PAID 状态下取消订单要退款，这两个动作挂到同一个cancel事件的不同转移上，代码看起来非常直观。

3.4 加一点“进入/离开动作”支持

手写版本最容易被吐槽的是缺少“进入状态时自动执行动作”。这个确实有用，比如每次进入 SHIPPED 都要发消息通知用户。要加也不难，在Transition里再加一个可选的enter_action，或者在状态机里维护一个state_action映射，进入新状态后统一执行。

我一般倾向于把“进入动作”和“转移动作”分开：转移动作处理事件本身的副作用，进入动作处理新状态带来的公共服务。比如发通知、记录审计日志、更新缓存，放在进入动作里会避免重复写。

4. 状态机真的越多越好吗：状态爆炸与分层时机

4.1 状态爆炸是真实存在的

状态机不是银弹。最典型的问题是“状态爆炸”：当你有两个维度互相独立的状态，比如“支付状态”和“物流状态”，硬塞进一个状态机，状态数量会变成笛卡尔积，转移表大得没法维护。

这时候我的第一反应不是换框架，而是拆状态机。支付状态机负责支付域，物流状态机负责物流域，两个状态机通过事件或消息驱动串联。不要试图用一个总状态机管所有事。

4.2 分层状态机和 HSM 什么时候才值得用

如果业务确实需要“子状态”概念，比如“发货中”下面还有“等待揽收”“运输中”“派送中”，可以考虑分层状态机（HSM）。但说实话，我见过的大部分团队在需要 HSM 之前，就已经被业务复杂度打败了，或者根本没有那么多共享转移。

我的判断标准是：如果共享转移超过三分之一，或者同一个事件在不同子状态下都要走几乎相同的逻辑，再考虑引入层级；否则，用扁平状态机加上复合状态命名，比如SHIPPING_WAIT_PICKUP、SHIPPING_IN_TRANSIT，也完全能撑住。不要因为“听起来高级”就上分层，分层状态机的调试复杂度是上了一个台阶的。

4.3 和“不变量”配合的测试思路

状态机最大的测试优势在于：你不需要覆盖所有可能输入，只需要覆盖转移表里的每一条边。每一条(state, event) -> new_state就是一个测试用例。我会为每条合法转移写一个用例，同时为若干条非法转移写“预期抛错”的用例，比如PENDING + ship必须报错。

这样测试数量是可控的，状态数乘以事件数，一个订单系统通常也就几十个用例。比起给十几个布尔组合写测试，状态机的测试既不重复又容易复盘。我在项目里还会再加一个不变量测试：不管怎么发事件，最终状态永远落在预定义的状态集合里，不会出现“半支付半发货”这种脏状态。

5. 我在真实项目里踩过的五个坑

5.1 异步事件的到达顺序

在线支付回调和前端轮询结果可能不是同一时刻到达的。如果支付已经成功，但先收到了用户点“取消”的事件，状态机就会在 PAID 状态处理 cancel，这时如果不做退款逻辑，就会产生“已取消但钱也扣了”的脏订单。

踩过这个坑之后，我的习惯是：事件里带上时间戳，在进入转移前先校验事件顺序；或者对于有外部回调的场景，把“取消”设计成只有 PENDING 和 PAID 能处理，并且 PAID 分支走退款。顺序校验错了，状态机反而会把脏数据合法化，这个比 if-else 更难查，因为它看起来每一步都是对的。

5.2 非法转移抛异常还是忽略

我一开始在send里对非法转移直接抛异常，结果线上偶尔出现一个ValueError。后来发现，有些事件是外部的、无序的、甚至重复的，比如支付回调可能推两遍。在这种情况下，纯抛异常会把业务整体打断。

现在的做法是分两层：如果是外部不可控事件，我定义一个IGNORED策略，重复事件直接忽略，并把日志记下来；如果是内部逻辑不该发生的事件，才抛异常。怎么区分？看业务容忍度。外部回调的重复事件通常跳过即可，内部 bug 导致的非法转移必须炸出来。

5.3 状态机与数据持久化的对齐

状态机跑在内存里很欢，但重启之后状态从哪来？大部分项目会把状态字段存到数据库，但要注意：你做转移判断时读到的状态，和落库时写下的状态，在分布式场景下可能已经被改掉了。所以我在写更新的时候，通常用条件更新：

UPDATE orders SET state = :new_state WHERE id = :id AND state = :old_state

如果影响行数为 0，说明并发修改了状态，这时候要回滚或者重试。这个和状态机本身无关，而是“状态字段也是数据”的意识。不加上这个，再好的状态机也会在并发下出现脏覆盖。

5.4 事件幂等不能靠状态机硬扛

有段时间我以为只要状态机定义得好，重复事件就进不来。现实是：消息队列可能重投，HTTP 回调可能重试，状态机只是告诉你“当前状态不接受这个事件”，但你没法区分“这个事件以前处理过”和“这个事件现在不该发生”。

我的做法是在事件处理函数里记录外部事件 ID，比如支付回调的 transaction_id，处理前先查一下去重表。如果已经处理过，直接返回成功。这件事不要交给状态机判断，否则状态迁移的语义会被幂等逻辑污染。

5.5 调试日志怎么打才有效

状态机的问题大多数发生在“事件到达时状态不对”。我现在的日志格式是：

[state_machine] order_id=123 state=PENDING event=pay_success -> PAID

同时把transitions表的内容在启动时打一条摘要，这样出问题时，我可以对着日志看出：当时状态是谁、事件是什么、转移走向哪里，而不是去翻一堆业务日志猜。日志里一定要带业务主键，不然分布式环境下面根本没法把事件串起来。

6. 一个我常用的简化技巧

6.1 用枚举而不是字符串

虽然前面的例子用了字符串，但正式项目里我强烈建议用枚举。字符串会拼错，还不好做 IDE 补全。用enum.Enum之后，转移表在启动时就能校验一遍：有没有引用不存在的状态、有没有漏掉某条路径。这个校验用几十行代码就能写，是手写状态机最大的红利。

from enum import Enum class OrderState(str, Enum): PENDING = "PENDING" PAID = "PAID" SHIPPED = "SHIPPED" COMPLETED = "COMPLETED" CANCELLED = "CANCELLED"

加上str混合之后，枚举值可以直接序列化到 JSON，数据库里存的还是字符串，但代码里的类型提示和自动补全都保住了。

6.2 把动作放到配置里

如果状态机被多个服务复用，可以把转移表变成一份 JSON 或 YAML 配置，动作名用字符串注册表映射。这样业务方改流程时只需要改配置，不用动代码。不过我不建议一上来就做这个抽象，等真正出现两份业务都用到同一套状态的时候再说，否则就是过度设计。

6.3 一个小工具函数

最后给一个我在调试时常用的函数：dump_machine，把当前状态机所有的转移关系打印成表格，用于代码评审和复盘：

def dump_machine(machine: SimpleStateMachine): for (state, event), trans in sorted(machine.transitions.items()): print(f"{state:20s} + {event:20s} -> {trans.target}")

你会在评审时发现，肉眼读转移表比读一堆 if-else 快得多。这也是我会在各种项目里反复推状态机的原因：它不是在约束你，而是把复杂逻辑变成一张可以讨论、可以评审、可以测试的表。把这张表打印出来贴在代码评审文档里，几乎每次都能揪出一两个大家都没想到的边界转移。