2. 事件驱动架构：事件驱动编程范式，观察者模式，事件循环与回调函数

好，咱们进入第二章。这一章聊的是事件驱动架构，说白了就是整个订单簿策略的「心脏」和「血管」。你想想看，行情数据一来，你得立刻响应，不能等。这就是事件驱动的核心思想——来了事，就办事。

我个人习惯把事件驱动比作「急诊室」。病人（事件）来了，分诊台（事件循环）立刻判断该找哪个科室（回调函数）。如果让病人排队等叫号，那黄花菜都凉了。量化交易里，毫秒级的延迟就是真金白银的损失。

2.1 事件驱动编程范式

传统的编程是「拉」模式——你主动去问数据到了没。事件驱动是「推」模式——数据到了，它主动通知你。这两种思维完全不同。

举个例子：

# 传统轮询方式（别这么干）
while True:
    orderbook = get_orderbook()  # 你一直在问
    if orderbook.has_update():
        process(orderbook)
    time.sleep(0.001)

# 事件驱动方式（推荐）
def on_orderbook_update(orderbook):
    process(orderbook)  # 数据来了，它叫你

event_engine.register('orderbook.update', on_orderbook_update)

我在项目中遇到过不少新手，上来就写死循环轮询。结果呢？CPU 跑满，延迟还高。后来改成事件驱动，CPU 占用率直接降到 5% 以下。说白了，别主动去问，等着被通知就行。

2.2 观察者模式

观察者模式，是事件驱动架构的「骨架」。它解决了一个核心问题：一个对象状态变化时，如何通知多个依赖它的对象？

你想想看，订单簿更新了，可能有十几个策略模块都需要知道。总不能每个模块都去轮询吧？观察者模式就是干这个的。

我画了一张图，帮你理解这个模式：

这张图你看懂了吗？订单簿是「主题」，它维护了一个观察者列表。一旦订单簿有更新，就调用 notify() 方法，遍历列表里的每个观察者，调用它们的 on_update()。这就是观察者模式的精髓。

代码实现其实很简单：

class OrderBookSubject:
    def __init__(self):
        self._observers = []  # 观察者列表
    
    def attach(self, observer):
        self._observers.append(observer)
    
    def detach(self, observer):
        self._observers.remove(observer)
    
    def notify(self, data):
        for obs in self._observers:
            obs.on_update(data)  # 逐个通知

class StrategyObserver:
    def on_update(self, orderbook_data):
        # 处理订单簿数据
        print(f"策略收到更新：{orderbook_data}")

2.3 事件循环与回调函数

事件循环，就是那个「一直在转的轮子」。它负责从事件队列里取事件，然后分发给对应的回调函数。回调函数，就是「你告诉我怎么做，我照做」。

我习惯把事件循环比作「邮递员」。他不停地从邮局（事件队列）拿信（事件），然后按地址（事件类型）送到各家各户（回调函数）。

一个典型的事件循环长这样：

import queue
import time

class EventEngine:
    def __init__(self):
        self._queue = queue.Queue()  # 事件队列
        self._handlers = {}          # 事件类型 -> 回调函数列表
        self._running = False
    
    def register(self, event_type, handler):
        """注册事件处理器"""
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)
    
    def put(self, event):
        """往队列里放事件"""
        self._queue.put(event)
    
    def _process(self, event):
        """处理单个事件"""
        handlers = self._handlers.get(event.type, [])
        for handler in handlers:
            try:
                handler(event.data)  # 调用回调函数
            except Exception as e:
                print(f"回调执行出错：{e}")
    
    def start(self):
        """启动事件循环"""
        self._running = True
        while self._running:
            try:
                # 阻塞等待，直到有事件到来
                event = self._queue.get(timeout=1)
                self._process(event)
            except queue.Empty:
                continue  # 超时继续循环
    
    def stop(self):
        self._running = False

嗯，这里要注意一个关键点：回调函数里不能做耗时操作。为什么？因为事件循环是单线程的，你一个回调卡住了，后面所有事件都得排队等着。

我在项目中遇到过这种情况：有个回调函数里做了网络请求，结果行情来了，它还在等网络响应。整个策略直接「冻住」了。后来怎么解决的？把耗时操作扔到线程池里：

from concurrent.futures import ThreadPoolExecutor

class SafeEventEngine(EventEngine):
    def __init__(self):
        super().__init__()
        self._executor = ThreadPoolExecutor(max_workers=4)
    
    def _process(self, event):
        handlers = self._handlers.get(event.type, [])
        for handler in handlers:
            # 耗时操作提交到线程池
            if self._is_heavy(handler):
                self._executor.submit(handler, event.data)
            else:
                handler(event.data)  # 轻量操作直接执行

2.4 实战：订单簿事件驱动框架

把上面这些串起来，就是一个完整的订单簿事件驱动框架。我直接给你看核心代码：

# 定义事件类型
class EventType:
    ORDERBOOK_SNAPSHOT = 'orderbook.snapshot'
    ORDERBOOK_UPDATE = 'orderbook.update'
    TRADE = 'trade'
    ORDER_STATUS = 'order.status'

# 定义事件对象
class Event:
    def __init__(self, type, data):
        self.type = type
        self.data = data
        self.timestamp = time.time()

# 订单簿策略基类
class OrderBookStrategy:
    def __init__(self, engine):
        self.engine = engine
        # 注册感兴趣的事件
        engine.register(EventType.ORDERBOOK_UPDATE, self.on_orderbook_update)
        engine.register(EventType.TRADE, self.on_trade)
    
    def on_orderbook_update(self, data):
        """子类实现具体的策略逻辑"""
        raise NotImplementedError
    
    def on_trade(self, data):
        """子类实现成交处理"""
        raise NotImplementedError

# 具体策略示例
class MyStrategy(OrderBookStrategy):
    def on_orderbook_update(self, data):
        # 检查价差，决定是否下单
        spread = data['ask_price'] - data['bid_price']
        if spread > 0.5:
            print(f"价差过大，准备套利：{spread}")
            # 这里可以调用下单接口
    
    def on_trade(self, data):
        print(f"成交了：{data}")

这个框架的好处是：策略逻辑和事件分发完全解耦。你想加一个新策略？继承 OrderBookStrategy，实现两个回调方法就行。其他代码一行都不用改。

组件	职责	我踩过的坑
事件引擎	管理事件队列和分发	队列用无界队列，内存爆过
事件对象	封装事件类型和数据	忘了加时间戳，回放时没法对齐
回调函数	处理具体业务逻辑	回调里抛异常，整个循环挂了
观察者列表	维护订阅关系	动态增删观察者导致迭代异常

我个人习惯在项目里用 asyncio 实现事件循环，性能更好。但今天讲的这个基础版本，足够你理解事件驱动的本质了。你想想看，是不是比轮询优雅多了？

---

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