第三章：事件循环优化——从基础while循环到高效的事件循环设计、事件优先级调度

事件循环，说白了就是回测引擎的“心脏”。

我刚开始做回测系统时，觉得事件循环不就是个while循环嘛，有什么好优化的？直到有一次，我的回测跑了一个小时还没跑完，而同样的策略在别人系统里只要十分钟。嗯，从那以后，我再也不敢小看这个“简单”的循环了。

3.1 基础while循环：能跑，但跑不快

最原始的事件循环长这样：

while True:
    event = get_next_event()
    if event is None:
        break
    process_event(event)

这段代码逻辑没错，但性能问题很大。我在项目中遇到过，当事件量达到百万级别时，这种写法会让CPU空转严重。为什么？因为get_next_event()如果返回None，循环就停了——但真实场景中，事件是分批到达的，中间会有空闲期。

更糟糕的是，如果事件处理耗时不均，一个慢事件会堵住后面所有事件。你想想看，一个订单成交事件处理了50毫秒，后面100个tick事件就只能排队等着。这合理吗？不合理。

核心问题：基础while循环缺乏对事件优先级、处理顺序、空闲管理的控制。说白了，它就是个“来什么处理什么”的被动模式。

3.2 高效事件循环设计：从被动到主动

高效的事件循环，核心思路是“主动调度”。我个人习惯把事件循环拆成三个层次：

  1. 事件源层：负责从不同渠道获取事件（文件、网络、内存队列）
  2. 调度层：决定下一个处理哪个事件
  3. 处理层：执行具体的事件逻辑

来看一个改进后的设计：

class EventLoop:
    def __init__(self):
        self._running = False
        self._event_queue = PriorityQueue()
        self._idle_handler = None
        
    def run(self):
        self._running = True
        while self._running:
            # 尝试获取事件，但设置超时
            try:
                event = self._event_queue.get(timeout=0.001)
                self._process_event(event)
            except Empty:
                # 空闲时执行低优先级任务
                if self._idle_handler:
                    self._idle_handler()
                    
    def _process_event(self, event):
        # 按事件类型分发
        handler = self._handlers.get(event.type)
        if handler:
            handler(event)

这里有几个关键改进：

  • 超时机制：避免死等，给空闲处理留出时间
  • 空闲处理：没有事件时，可以做数据整理、内存回收
  • 事件分发：按类型路由，避免大函数

我的经验：超时时间设多少合适？我一般设0.5-1毫秒。太短会导致频繁轮询，浪费CPU；太长会让事件响应变慢。你可以用性能分析工具测一下，找到平衡点。

3.3 事件优先级调度：让重要的事先办

回测中，不是所有事件都同等重要。举个例子：

  • 高优先级：订单成交、风控信号（延迟会影响策略逻辑）
  • 中优先级：tick数据更新、K线生成
  • 低优先级：日志写入、统计计算、UI刷新

我曾经犯过一个错误：把所有事件都放在同一个队列里，结果一个日志写入操作卡住了订单处理，导致回测结果出现偏差。那次教训让我意识到，优先级调度不是“锦上添花”，而是“刚需”。

实现优先级调度，我推荐两种方式：

方式一：多队列 + 轮询

class PriorityEventLoop:
    def __init__(self):
        self._queues = {
            'high': deque(),
            'medium': deque(),
            'low': deque()
        }
        
    def _get_next_event(self):
        # 高优先级队列优先
        for priority in ['high', 'medium', 'low']:
            if self._queues[priority]:
                return self._queues[priority].popleft()
        return None

方式二：带权重的优先级队列

import heapq

class WeightedEventLoop:
    def __init__(self):
        self._heap = []
        self._counter = 0
        
    def add_event(self, event, priority):
        # 优先级数值越小，越先处理
        heapq.heappush(self._heap, (priority, self._counter, event))
        self._counter += 1
        
    def _get_next_event(self):
        if self._heap:
            _, _, event = heapq.heappop(self._heap)
            return event
        return None

注意：使用优先级队列时，要小心“饥饿”问题。如果高优先级事件源源不断，低优先级事件可能永远得不到处理。我通常会给低优先级事件设置一个“最大等待时间”，超时后强制提升优先级。

3.4 事件循环的架构图

下面这张图展示了我个人比较推崇的事件循环架构：

高效事件循环架构图 事件源层 文件事件 网络事件 内存队列事件 定时事件 调度层 高优先级队列 订单成交、风控 中优先级队列 tick、K线 低优先级队列 日志、统计 调度器：按优先级 + 防饥饿策略 处理层 事件分发器 策略处理器 风控处理器 日志处理器

这张图里，事件从源层进入，经过调度层的优先级排序，最终到达处理层。每个层次职责清晰，互不干扰。我个人觉得，这种分层设计最大的好处是——你想优化哪一层，就改哪一层，不会牵一发动全身。

3.5 实战中的避坑指南

讲几个我踩过的坑：

  • 坑一：事件处理中不要阻塞。我曾经在事件处理函数里直接调用了数据库写入，结果I/O等待把整个循环卡死了。解决方案：异步处理或使用线程池。
  • 坑二：小心事件积压。如果事件生产速度大于消费速度，内存会暴涨。我一般会加一个“背压”机制——当队列长度超过阈值时，主动丢弃低优先级事件。
  • 坑三：优先级反转。低优先级事件持有高优先级事件需要的锁，导致高优先级事件被阻塞。解决办法：使用无锁数据结构，或者给每个优先级分配独立的资源。

性能调优口诀：事件循环要高效，分层调度不能少。优先级里防饥饿，空闲时刻别空耗。

好了，这一章的内容就到这里。事件循环优化是个系统工程，从基础while循环到优先级调度，每一步都有讲究。希望你能从我的经验里找到适合自己的方案。

公众号:蓝海数据掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →