2、事件驱动架构：事件驱动编程模型、观察者模式、事件循环与回调、异步处理基础

好，咱们进入第二章。这一章我打算聊聊事件驱动架构。说实话，这是整个订单簿交易框架的灵魂。你想想看，市场数据每秒成千上万条涌进来，如果还用那种「你问我答」的同步方式，系统早就卡死了。

事件驱动，说白了就是「有事您说话，没事别烦我」。系统里各个模块各自待命，一旦有事件发生——比如新订单来了、价格变了——就立刻响应。这种模式特别适合高频交易场景。

2.1 事件驱动编程模型

事件驱动编程，核心就三个角色：事件源、事件通道、事件处理器。

  • 事件源：谁产生事件？交易所的行情推送、用户的下单指令、风控模块的告警……这些都是事件源。
  • 事件通道：事件怎么传递？通常是一个队列或者消息总线。我习惯用 Python 的 queue.Queue 或者 Redis 的 List。
  • 事件处理器：谁来处理？各个策略模块、风控模块、日志模块，它们订阅自己感兴趣的事件。

举个例子，一个简单的订单簿更新事件：

class OrderBookEvent:
    def __init__(self, event_type, data):
        self.event_type = event_type  # 'ADD', 'UPDATE', 'DELETE'
        self.data = data              # 订单详情
        self.timestamp = time.time()  # 事件发生时间

嗯，这里要注意：事件本身要尽量轻量。我在项目中见过有人把整个订单簿对象塞进事件里，结果内存直接爆了。事件只传递必要信息，比如订单ID、价格、数量，就够了。

2.2 观察者模式

观察者模式，是事件驱动最经典的实现方式。它定义了一种一对多的依赖关系：一个主题对象（Subject）可以通知多个观察者（Observer）。

我刚开始做量化时，把观察者模式理解成了「广播」。其实不对。广播是无差别的，而观察者模式是有选择地通知。每个观察者只关心自己订阅的事件类型。

来看一个精简的实现：

class EventBus:
    def __init__(self):
        self._subscribers = {}  # 事件类型 -> [处理器列表]

    def subscribe(self, event_type, handler):
        if event_type not in self._subscribers:
            self._subscribers[event_type] = []
        self._subscribers[event_type].append(handler)

    def publish(self, event):
        handlers = self._subscribers.get(event.event_type, [])
        for handler in handlers:
            handler(event)

我曾经踩过一个坑：订阅了事件，但忘记取消订阅。结果系统运行几天后，内存里堆积了成千上万个无用的处理器引用。所以，一定要在模块销毁时取消订阅。

避坑指南：我曾经在回测系统里用观察者模式，结果每个策略都订阅了「全部成交」事件。后来发现，一个订单成交，所有策略都去处理，导致重复计算。解决方案是：在事件里加上策略ID，处理器先判断是否与自己相关。

2.3 事件循环与回调

事件循环，就是那个「一直在跑、不断检查有没有新事件」的循环。它通常长这样：

while True:
    event = event_queue.get()  # 阻塞等待事件
    handle_event(event)        # 处理事件

你可能会问：这不就是个死循环吗？对，但它不是普通的死循环。它是有调度策略的。比如，高优先级的事件先处理，或者某些事件可以合并处理。

回调函数，是事件循环的「搭档」。当事件发生时，事件循环会调用对应的回调函数。回调函数不能阻塞，否则整个事件循环就卡住了。

我记得有一次，一个同事在回调函数里做了数据库查询，结果查询耗时 200 毫秒。在交易系统里，200 毫秒足够错过好几个 tick 了。所以，回调函数里绝对不能做耗时操作。

我的习惯：回调函数只做三件事——更新状态、触发新事件、记录日志。其他所有耗时操作，都丢到异步任务池里。

2.4 异步处理基础

异步处理，是为了解决「一个事件处理太久，影响其他事件」的问题。Python 里常用的异步方式有：

  • 多线程：每个事件处理器跑在独立线程里。但要注意 GIL 锁。
  • 协程：用 asyncio 实现，适合 I/O 密集型任务。
  • 进程池：适合 CPU 密集型计算，比如策略信号生成。

我个人推荐在订单簿系统里用协程 + 线程池的组合。协程处理网络 I/O（比如接收行情），线程池处理计算任务（比如更新订单簿）。

来看一个简单的异步事件处理器：

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def handle_order_book_update(event):
    # 异步处理网络 I/O
    data = await fetch_additional_data(event)
    # 同步计算任务丢到线程池
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, heavy_computation, data)
    # 发布结果事件
    event_bus.publish(ResultEvent(result))

这里有个关键点：事件处理不能保证顺序。如果你依赖事件顺序（比如先加订单再删订单），就必须在事件里加上序列号，或者用有界队列保证顺序。

核心总结：事件驱动架构让系统变得松耦合、高内聚。每个模块只关心自己订阅的事件，互不干扰。但代价是调试困难——事件流在系统里乱窜，你很难追踪一个请求的完整路径。所以，一定要做好日志和链路追踪。

2.5 本章知识体系

下面这张图，是我自己画的事件驱动架构核心逻辑。你看一眼，基本就明白整个体系怎么串起来的。

事件驱动架构核心逻辑 事件源 行情推送 / 用户指令 / 风控告警 事件通道 队列 / 消息总线 / Redis 事件循环 while True: get() → handle() 观察者模式（EventBus） subscribe(event_type, handler) → publish(event) → 通知所有订阅者 每个处理器只处理自己订阅的事件类型，避免无效唤醒 异步处理 协程（I/O） + 线程池（计算） + 进程池（CPU密集型） 回调函数不能阻塞！耗时操作必须异步化

这张图从左到右、从上到下，完整展示了事件从产生到最终处理的流程。你写代码时，就按这个结构来组织模块，基本不会出错。

公众号:蓝海资料掘金营，微信deep3321

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