第 2 章 · 01 EventEngine 的初始化与生命周期

第 2 章 · 01 EventEngine 的初始化与生命周期

本节摘要:本节精读 vnpy/event/engine.py 的骨架部分——Event 数据结构、EventEngine 的 __init__ 初始化、_run 事件主循环、_run_timer 定时器线程、start/stop 生命周期。你会看到一个完整的发布订阅总线怎么用 100 行搭起来:两条后台线程(主循环 + 定时器)配合一个线程安全 Queue,通过 _active 标志位优雅启停。读完本节,你理解了 VeighNa 所有数据流"为什么是异步的、为什么是顺序的、为什么不会丢事件"。

内容来源:原项目源码 vnpy/event/engine.py(共 146 行),精读并套用体系化模板。

学习目标

阅读完本节,你应当能够:

1. 逐行读懂 EventEngine 的 __init__ 六个成员各自的作用。
2. 解释 _run 主循环为什么 queue.get(timeout=1) 而非无限阻塞。
3. 说清 _run_timer 定时器线程每秒做什么。
4. 理解 start/stop 的标志位 + join优雅退出机制。
5. 对比 Queue+Thread 与 asyncio,说清 VeighNa 为什么选前者。

一、模块导入与常量

文件开头(engine.py:5-13):

5   from collections import defaultdict
6   from collections.abc import Callable
7   from queue import Empty, Queue
8   from threading import Thread
9   from time import sleep
10  from typing import Any
13  EVENT_TIMER = "eTimer"
​

每个导入都有用意:

• defaultdict:handlers 字典的默认工厂,键不存在时自动建空 list。
• Queue(queue.Queue):线程安全 FIFO 队列,连接"生产者"和"消费者"线程。
• Thread:用两条后台线程分别跑事件主循环和定时器。
• Empty:queue.get 超时时抛的异常,主循环靠它感知"队列空了"。
• EVENT_TIMER = "eTimer":定时器事件的固定 type 字符串,是引擎内置唯一的事件类型常量;其它业务事件(行情/订单)的 type 由各 gateway/策略自定义。

二、Event 数据结构

engine.py:16-27:

16  class Event:
23      def __init__(self, type: str, data: Any = None) -> None:
24          """"""
25          self.type: str = type
26          self.data: Any = data
​

极简——只有两个字段:

• type: str:事件类型字符串,引擎据此分发(如 "eTick."、"eOrder.")。
• data: Any:载荷,任意类型(通常是 TickData/OrderData 等 dataclass,第 3 章详讲)。

💡 核心心法:Event 故意用普通类而非 dataclass。原因是事件被高频创建/分发(每秒成百上千个 tick),避开 dataclass 的 __init__ 开销;且 Event 只有两个字段,手写 __init__ 比 dataclass 更直观。这是性能与简洁的权衡。

紧跟着的类型别名(engine.py:30):

30  HandlerType = Callable[[Event], None]
​

定义"处理器"的统一签名:接收一个 Event,无返回值。所有 register 方法的形参都用它。

三、EventEngine 的 init

核心初始化(engine.py:42-53):

42      def __init__(self, interval: int = 1) -> None:
47          self._interval: int = interval              # 定时器间隔(秒),默认 1s
48          self._queue: Queue = Queue()                # 事件队列(生产/消费解耦)
49          self._active: bool = False                  # 引擎运行标志,控制两个线程循环
50          self._thread: Thread = Thread(target=self._run)        # 事件主循环线程
51          self._timer: Thread = Thread(target=self._run_timer)  # 定时器线程
52          self._handlers: defaultdict = defaultdict(list)        # {type: [handler,...]}
53          self._general_handlers: list = []            # 通用处理器,监听所有事件
​

六个成员,全部以 _ 开头表示私有:

注意:Thread 对象在构造时就绑定了 target(如 Thread(target=self._run)),但尚未启动——直到 start() 才真正 thread.start()。这是"构造 ≠ 启动"的常见 Python 多线程写法。

四、_run 事件主循环

engine.py:55-64:

55      def _run(self) -> None:
59          while self._active:
60              try:
61                  event: Event = self._queue.get(block=True, timeout=1)
62                  self._process(event)
63              except Empty:
64                  pass
​

四行核心,信息量很大:

1. while self._active:死循环,靠标志位优雅退出。stop 时把 _active=False,循环自然结束。
2. queue.get(block=True, timeout=1):阻塞最多 1 秒拉取事件。
3. timeout=1 是关键:如果用无限阻塞(timeout=None),那么 stop 设了 _active=False 后,主循环还在 get 里卡着,根本不会重新检查 _active——线程永远退不出。加 1 秒超时,保证即使队列空,每秒也会醒来一次重新检查标志位。
4. 捕获 Empty:超时没拿到事件,什么都不做,继续循环。

💡 核心心法:timeout=1 不是性能考量,而是优雅退出的必要条件。这是 Python 多线程"标志位 + 阻塞队列"模式的标准写法——阻塞调用必须带超时,否则无法响应停机信号。

五、_run_timer 定时器线程

engine.py:80-87:

80      def _run_timer(self) -> None:
84          while self._active:
85              sleep(self._interval)
86              event: Event = Event(EVENT_TIMER)
87              self.put(event)
​

每 interval 秒(默认 1 秒)向队列投递一个 EVENT_TIMER 事件。这是引擎内置的心跳,策略层常用它做:

• 定期撤单检查(挂单超过 N 秒未成交则撤);
• 定时下单(每天 9:30 下班单);
• 统计指标刷新(每秒重算 RSI/MACD);
• GUI 时钟更新。

注意 sleep 在循环开头,所以启动后先等一个 interval 才发出第一个 tick。

六、start / stop 生命周期

engine.py:89-103:

89      def start(self) -> None:
93          self._active = True
94          self._thread.start()
95          self._timer.start()

97      def stop(self) -> None:
101         self._active = False
102         self._timer.join()
103         self._thread.join()
​

• start:先置 _active=True,再启动两个线程(顺序重要——先翻标志再起线程,避免线程起来时标志还是 False 立刻退出)。
• stop:先翻标志,再 join 等待两个线程退出。因为 _run 用了 timeout=1、_run_timer 用了 sleep(interval),最长约 1 秒内两个线程都会自然退出。

⚠️ 重要说明:stop 时 Queue 和已注册的 handlers 不会被清空。但 Python 的 Thread 对象不能再次 start(线程对象是一次性的),所以实际使用中 EventEngine 一般是单例长期运行(MainEngine 持有它直到 close),不会反复 start/stop。如果你要重启,得 new 一个新的 EventEngine。

七、为什么选 Queue+Thread 而非 asyncio

很多人会问:Python 有 asyncio,VeighNa 为什么不用协程?四个原因:

1. 对接的接口是同步阻塞 + 回调线程模型。CTP/期货/股票的 C++ API 都是同步调用 + 回调线程,Queue/Thread 与之天然贴合;asyncio 反而要桥接同步接口,复杂度更高。
2. GUI 也是事件循环模型。Qt 自己有事件循环,混 asyncio 进来要 qasync 这类胶水库,不稳。VeighNa 的 UI 刷新靠"事件总线 → Qt signal"中转,Queue 模式更简单。
3. Queue 跨线程安全且简单。queue.Queue 内部有锁,是 Python 最成熟的跨线程通道;asyncio 要求全程 async,生态(数据库驱动/SDK)未必都 async。
4. 100 行能讲清。整个 EventEngine 用最朴素的标准库实现,可读性极高,降低贡献门槛。

💡 核心心法:技术选型要看上下文。VeighNa 要对接一堆 C++ 同步接口 + Qt GUI,asyncio 的优势(高并发 IO)用不上,反而增加桥接成本。Queue+Thread 是"够用且最稳"的选择。这是务实的工程判断,不是落后。

本节要点回顾

1. Event 故意用普通类(非 dataclass),追求高频场景的轻量。
2. init 六成员:queue / active / thread / timer / handlers / general_handlers。
3. _run 主循环:while _active + get(timeout=1) + 捕获 Empty,timeout 是优雅退出的关键。
4. _run_timer:每 interval 秒 put 一个 EVENT_TIMER 心跳,供策略做定时任务。
5. start/stop:标志位 + join,Thread 对象一次性不能重启,故 EventEngine 单例长跑。
6. 选 Queue+Thread 而非 asyncio:贴合 C++ 同步接口 + Qt GUI,务实选择。

下一节,我们看事件怎么被投递和分发——put / _process / register 三大机制,以及"专属处理器"和"通用处理器"的分工。