快速开始
本页的每一段代码都在 Python 3.13 上实际执行验证过,输出为真实输出。
安装
pip install zoo-framework需要 Python 3.13 及以上。验证:
python -c "import zoo_framework; print(zoo_framework.__version__)"第一个任务:手写一个最小示例
先用单文件跑通,理解最小闭环。把下面存成 main.py:
from zoo_framework.core import Master
from zoo_framework.workers import BaseWorker
class HelloWorker(BaseWorker):
def __init__(self):
super().__init__({
"is_loop": True, # 跨调度轮次持续运行
"delay_time": 1.0, # 每轮执行后等待的秒数
"name": "HelloWorker",
})
self.counter = 0
def _execute(self):
self.counter += 1
print(f"[HelloWorker] tick #{self.counter}", flush=True)
if __name__ == "__main__":
master = Master()
master.register_worker("HelloWorker", HelloWorker) # 注册的是**类**
master.run()运行:
python -u main.py真实输出(注意:默认会打印框架自身的调度日志,见下节如何关掉):
[HelloWorker] tick #1
[HelloWorker] tick #2
[HelloWorker] tick #3
...按 Ctrl-C 停止。
两个容易踩的地方
register_worker传的是类,不是实例。WorkerRegistry用issubclass校验, 传函数或实例会得到TypeError: issubclass() arg 1 must be a class。- 管道/文件输出时加
-u。 否则 Python 会块缓冲print,看不到实时输出。
把日志调安静
不配置的话,框架会在控制台打印每个 Worker 每一轮的启停日志,你的输出会被淹掉。
在工作目录放一个 config.json:
{
"log": { "path": "./logs", "level": "warning" }
}实测对比(同样运行 5 秒):
| 配置 | 总输出行数 | 其中属于你的输出 |
|---|---|---|
| 默认 | 23 | 3 |
log.level: warning | 3 | 3 |
默认级别偏高是一个已知问题,在 issue #111 中跟踪。
用脚手架生成项目结构
单文件够验证,但真实项目需要目录结构:
zfc --create my_app
cd my_app产出:
my_app/
├── config.json
└── src/
├── main.py # 入口:注册 Worker 并启动
├── conf/ # 启动期配置钩子(在 Master() 构造时执行)
├── params/ # 配置项声明(对应 config.json 的 demo 段)
├── events/ # 事件反应器
└── workers/ # 任务单元再生成一个任务:
zfc --worker order_sync脚手架的两个已知限制
--worker会把 import 与注册写进src/main.py的WORKERS列表(不是写进workers/__init__.py)。--worker order_sync目前生成的类名是Order_SyncWorker(下划线被保留), 而不是OrderSyncWorker。功能正常,命名修复在 issue #112 中跟踪。--create生成的WORKERS列表初始为空,不跑一次--worker就看不到任何业务输出 (issue #110)。
运行:
python -u src/main.py接下来
| 我想… | 去哪里 |
|---|---|
| 让状态在重启后恢复 | 状态机 |
| 让两个任务互相通信 | 事件系统 |
| 跨任务共享实例 | 作用域容器 |
| 查签名与参数 | API 参考 |
常见问题
TypeError: issubclass() arg 1 must be a class
master.register_worker("MyWorker", MyWorker()) # ✗ 传了实例
master.register_worker("MyWorker", lambda: MyWorker()) # ✗ 传了工厂函数
master.register_worker("MyWorker", MyWorker) # ✓ 传类看不到任何输出
依次排查:加 python -u;确认 log.level 没被设成 error 以上; 确认注册了 Worker 再调用 master.run()——Master() 单独构造只会跑两个内建系统 Worker。
AttributeError: property 'is_loop' of ... has no setter
is_loop 是只读属性,唯一真源是构造时传入的 props。不要写 self.is_loop = True, 要在 super().__init__({...}) 的字典里声明。
ModuleNotFoundError: No module named 'zoo_framework'
装到了别的解释器。用 python -m pip install zoo-framework 确保与运行时同一个解释器。
本页内容的可靠性
本页不再包含手工编写的"预期输出"。所有代码块与输出均在 Python 3.13 上实际运行得到; 文中出现的每一处 API 都以 zoo_framework 的实际导出面为准。
背景:在本页改写之前,它包含数个并不存在的 API (
EventChannelRegister、StateMachineManager.create_state_machine/add_state/transfer), 以及一段没有register_worker的入口代码——照做的话程序什么都不会运行。 这类错误在文字上是看不出来的,只能靠"跑一遍"发现。