状态机设计与实现

章节：进阶篇第5章
难度：⭐⭐⭐ 进阶级
阅读时间：25-30 分钟
核心收获：掌握状态机设计方法，学会用状态机建模自动化任务

8.1 理论要点：状态机的核心概念

8.1.1 什么是状态机

状态机（State Machine） 是一种建模方法，将系统的行为抽象为： - 状态（State）：系统在某一时刻的处境 - 转移（Transition）：从一个状态到另一个状态的变化 - 触发条件（Trigger）：导致状态转移的事件 - 动作（Action）：状态转移时执行的操作

8.1.2 为什么自动化任务需要状态机

问题：自动化任务在真实环境中会遇到各种异常： - 数据源超时 - API 限流 - 推送失败 - 内容不足

状态机的价值： - 把"如果...怎么办"显式化 - 每个状态有明确的成功出口和失败出口 - 避免"面条代码"式的异常处理 - 易于测试和验证

8.1.3 状态机的核心原则

8.2 实战案例：热点选题任务的状态机

8.2.1 完整状态流转图

stateDiagram-v2
    [*] --> Idle
    Idle --> Triggered: 定时/手动触发
    Triggered --> SourceCheck: 开始运行
    SourceCheck --> Fetching: 数据源就绪
    SourceCheck --> Blocked: 数据源不可用（≥2个失败）
    Fetching --> Aggregating: 全部响应
    Fetching --> PartialAggregating: 部分超时
    Aggregating --> Filtering: 聚合完成
    PartialAggregating --> Filtering: 标记缺失后继续
    Filtering --> Delivering: 有效条目≥5条
    Filtering --> Blocked: 有效条目不足
    Delivering --> Completed: 推送成功
    Delivering --> Blocked: 推送失败
    Blocked --> Idle: 告警后等待下次触发
    Completed --> Idle: 完成

8.2.2 状态定义详解

8.2.3 状态文件设计

{
  "batch_id": "ai-hotspot-2026-07-10",
  "trigger_time": "2026-07-10T09:00:00+08:00",
  "state": "delivering",
  "completed": ["fetching", "aggregating", "filtering"],
  "source_status": {
    "wechat": "ok",
    "github": "ok",
    "multi_search": "ok",
    "aihot": "ok"
  },
  "item_count": 18,
  "last_error": null,
  "updated_at": "2026-07-10T09:02:14+08:00"
}

8.3 实现方式对比

8.3.1 三种实现方式

8.3.2 Python 实现示例

from enum import Enum
from dataclasses import dataclass
from datetime import datetime

class State(Enum):
    IDLE = "idle"
    TRIGGERED = "triggered"
    SOURCE_CHECK = "source_check"
    FETCHING = "fetching"
    PARTIAL_AGGREGATING = "partial_aggregating"
    AGGREGATING = "aggregating"
    FILTERING = "filtering"
    DELIVERING = "delivering"
    COMPLETED = "completed"
    BLOCKED = "blocked"

@dataclass
class TaskState:
    batch_id: str
    state: State
    completed: list
    source_status: dict
    item_count: int
    last_error: str = None
    updated_at: str = None

class StateMachine:
    def __init__(self, config):
        self.config = config
        self.state = TaskState(
            batch_id=self.generate_batch_id(),
            state=State.IDLE,
            completed=[],
            source_status={},
            item_count=0,
            updated_at=datetime.now().isoformat()
        )

    def transition(self, event):
        """状态转移"""
        current = self.state.state

        # 定义状态转移规则
        transitions = {
            State.IDLE: {
                "trigger": State.TRIGGERED
            },
            State.TRIGGERED: {
                "source_ready": State.SOURCE_CHECK,
                "source_not_ready": State.BLOCKED
            },
            State.SOURCE_CHECK: {
                "ready": State.FETCHING,
                "not_ready": State.BLOCKED
            },
            State.FETCHING: {
                "all_ok": State.AGGREGATING,
                "partial_timeout": State.PARTIAL_AGGREGATING
            },
            State.PARTIAL_AGGREGATING: {
                "continue": State.FILTERING,
                "all_failed": State.BLOCKED
            },
            State.AGGREGATING: {
                "done": State.FILTERING,
                "error": State.BLOCKED
            },
            State.FILTERING: {
                "pass": State.DELIVERING,
                "blocked": State.BLOCKED
            },
            State.DELIVERING: {
                "success": State.COMPLETED,
                "failed": State.BLOCKED
            },
            State.BLOCKED: {
                "retry": State.IDLE,
                "abort": State.IDLE
            },
            State.COMPLETED: {
                "reset": State.IDLE
            }
        }

        # 执行转移
        if event in transitions.get(current, {}):
            new_state = transitions[current][event]
            self.state.state = new_state
            self.state.updated_at = datetime.now().isoformat()
            return True
        return False

8.4 图文说明

📌 待补充：建议绘制以下示意图 1. 状态机完整流程图（10个状态 + 转移箭头） 2. 状态文件结构图 3. 状态转移决策树 4. Python 代码执行流程图

8.5 常见错误

📝 版本迭代记录