技术	版本	选择理由
FastAPI	>= 0.109.0	高性能、自动生成 API 文档、类型安全
Uvicorn	>= 0.27.0	高性能 ASGI 服务器
python-socketio	>= 5.11.0	成熟的 WebSocket 解决方案

配置管理

技术	版本	选择理由
pydantic-settings	>= 2.1.0	类型安全、支持环境变量、.env 文件

调度器

技术	版本	选择理由
APScheduler	>= 3.10.0	成熟的定时任务调度库、支持 Cron 表达式

浏览器自动化

技术	版本	选择理由
DrissionPage	>= 4.1.0	功能强大、易于使用、支持 Chromium

缓存和队列

技术	版本	选择理由
Redis	>= 5.0.0	高性能、支持多种数据结构、分布式

数据库

技术	版本	选择理由
aiomysql	>= 0.2.0	异步 MySQL 客户端、高性能

设计模式

1. 装饰器模式

用于工作流和步骤的声明式注册：

@wf.register_class()
class MyWorkFlow(StepActionHandler):
    @wf.step(sort=1)
    def step_one(self):
        return self.set_success()

2. 生成器模式

用于步骤的顺序执行：

def _get_step_generator(self, step_action_list, *args, **kwargs):
    for step_name in step_action_list:
        # 执行步骤
        result = self._execute_step(step_name)
        yield result

3. 模板方法模式

StepActionHandler 定义了工作流执行的骨架：

def run_all_steps(self, *args, **kwargs):
    # 模板方法
    for step_result in self._get_step_generator(step_action_list):
        # 处理每个步骤的结果
        pass

4. 观察者模式

生命周期钩子系统基于观察者模式：

# HookManager 是主题
class HookManager:
    _hooks: Dict[HookEvent, List[HookCallback]] = {}

    @classmethod
    def register(cls, event, callback):
        # 注册观察者
        pass

    @classmethod
    def emit(cls, event, **kwargs):
        # 通知所有观察者
        pass

5. 单例模式

GlobalDataManager 使用类变量实现单例：

class GlobalDataManager:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

6. 策略模式

不同的执行模式对应不同的策略：

class WorkDataHandler:
    def start_workflow(self, cls):
        mode = cls.runtype.get("mode")
        if mode == "single":
            return self._run_single(cls)
        elif mode == "traversal":
            return self._run_traversal(cls)
        elif mode == "indefinitely":
            return self._run_indefinitely(cls)
        # ...

线程安全

全局数据管理器

class GlobalDataManager:
    def __init__(self):
        self._data = {}
        self._lock = threading.Lock()

    def get(self, key, default=None):
        with self._lock:
            return self._data.get(key, default)

    def set(self, key, value):
        with self._lock:
            self._data[key] = value

线程控制器

class ThreadController:
    def __init__(self):
        self._pause_event = threading.Event()
        self._stop_event = threading.Event()

    def pause(self):
        self._pause_event.clear()

    def resume(self):
        self._pause_event.set()

    def stop(self):
        self._stop_event.set()

扩展性

自定义执行模式

def custom_logic(self, cls, **kwargs):
    """自定义执行逻辑"""
    # 实现自定义逻辑
    pass

manager.work_data_handler.set_custom_work(custom_logic)

自定义钩子

from hsxworkflow import HookManager, HookEvent

HookManager.register(HookEvent.WORKFLOW_START, my_callback)

自定义状态枚举

from hsxworkflow import StepStatus

MyStatus = StepStatus.set_new_status("CUSTOM_STATUS", 200)

性能优化

1. 线程池管理

使用 ThreadPoolExecutor 实现并发执行：

self.th_work = ThreadPoolExecutor(max_workers=max_workers)

2. 异步 I/O

使用 FastAPI 和 aiomysql 实现异步 I/O：

async def get_workflow_detail(class_id: int):
    info = await database.get_workflow(class_id)
    return info

3. 缓存机制

使用 Redis 缓存频繁访问的数据：

rdb = get_redis_handler()
cached_data = rdb.get("workflow_cache")

4. WebSocket 推送优化

使用房间机制实现精准推送：

# 只向特定房间推送
await sio.emit(event, data, room=str(class_id))

下一步

现在你已经了解了 HsxWorkFlow 的技术架构，可以继续学习：

核心引擎 - 深入了解核心引擎的实现
执行模式 - 探索不同的执行模式
高级特性 - 学习高级功能的使用