第2章核心组件设计

核心组件设计：事件总线、事件处理器、事件队列、事件路由

好，咱们接着聊。上一章我们把事件驱动架构的骨架搭起来了，现在该往里面填「器官」了。

核心组件就四个：事件总线、事件处理器、事件队列、事件路由。说白了，它们就是整个系统的「血管」、「心脏」、「血库」和「调度员」。我当年第一次设计这套东西时，把事件总线和路由混在一起写，结果调试了三天三夜……嗯，从那以后我再也不敢偷懒了。

1. 事件总线（Event Bus）

事件总线是什么？它就是所有事件的「高速公路」。任何组件想发事件，都走这条路；任何组件想收事件，也在这条路上等着。

我个人习惯把事件总线设计成单例模式。为什么？因为全局只需要一条总线，多了反而乱。你想想看，如果每个模块都有自己的总线，那事件怎么跨模块传递？

核心职责：

提供统一的发布接口（publish / emit）
提供统一的订阅接口（subscribe / on）
管理订阅者列表
事件过滤与分发

代码示例（Python 伪代码）：

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.type, [])
        for handler in handlers:
            handler(event)

这里有个坑：订阅者泄漏。我曾经在一个高频交易系统里，忘了在模块销毁时取消订阅，结果事件越积越多，内存直接爆了。所以，记得加一个 unsubscribe 方法。

注意：事件总线不要做太多事情。它只负责「传递」，不负责「处理」。处理逻辑交给事件处理器。

2. 事件处理器（Event Handler）

事件处理器就是干活的。它收到事件后，执行具体的业务逻辑——比如更新订单状态、计算风险指标、发送报警通知。

我建议把处理器设计成接口/抽象类，这样每个处理器只关注自己那一亩三分地。举个例子：

class EventHandler(ABC):
    @abstractmethod
    def handle(self, event):
        pass

class OrderFilledHandler(EventHandler):
    def handle(self, event):
        # 更新订单状态
        # 计算成交均价
        # 触发风控检查
        pass

为什么要这样？因为量化交易系统里，事件类型太多了。行情事件、订单事件、风控事件、账户事件……如果全写在一个大函数里，那代码就是一团乱麻。我见过一个项目，一个 handler 函数写了 2000 行，后来没人敢改。

小技巧：给每个处理器加一个 can_handle 方法，用来判断当前事件是否该由它处理。这样路由逻辑就清晰了。

3. 事件队列（Event Queue）

事件队列是系统的「缓冲池」。为什么需要它？因为事件的生产速度和消费速度往往不一致。

比如行情来了，一秒 1000 笔 tick，但你的策略计算可能一秒只能处理 200 笔。如果没有队列，事件直接丢给处理器，系统就崩了。

我个人习惯用优先级队列。为什么？因为不是所有事件都同等重要。举个例子：

高优先级：风控事件、撤单事件（必须立即处理）
中优先级：订单成交事件
低优先级：行情 tick、日志事件

代码示例：

import heapq

class PriorityEventQueue:
    def __init__(self):
        self._queue = []
        self._index = 0

    def put(self, event, priority=0):
        heapq.heappush(self._queue, (priority, self._index, event))
        self._index += 1

    def get(self):
        return heapq.heappop(self._queue)[-1]

注意：队列不能无限增长。一定要设置最大长度，或者用「丢弃策略」。我见过一个系统，队列满了之后直接 OOM，整个交易终端挂了 10 分钟……那损失，你懂的。

4. 事件路由（Event Router）

事件路由是系统的「交通指挥员」。它决定：这个事件该发给谁？

最简单的路由就是「根据事件类型分发」。但实际项目中，往往需要更复杂的规则：

基于内容的路由：比如「价格超过 100 的 tick 事件发给策略 A，低于 100 的发给策略 B」
基于来源的路由：比如「来自交易所 A 的事件走通道 1，来自交易所 B 的走通道 2」
基于负载的路由：比如「轮询发给多个处理器，均衡负载」

我建议把路由逻辑单独抽出来，不要和总线混在一起。这样后期改路由规则时，不用动总线代码。

class EventRouter:
    def __init__(self):
        self._routes = []  # [(条件, 处理器)]

    def add_route(self, condition, handler):
        self._routes.append((condition, handler))

    def route(self, event):
        for condition, handler in self._routes:
            if condition(event):
                handler.handle(event)

核心原则：路由只做「判断和转发」，不做「处理」。处理是处理器的事。

5. 整体架构图

下面我用一张 SVG 图把这四个组件的关系画出来。你一看就明白了。

6. 组件协作流程

这四个组件是怎么配合的？我拿一个「订单成交事件」举个例子：

事件源（交易所接口）收到成交回报，生成 OrderFilledEvent
调用 事件总线 的 publish 方法，把事件丢进去
总线把事件交给 事件路由，路由根据事件类型匹配到 OrderFilledHandler
路由把事件放入 事件队列，等待处理
事件处理器 从队列中取出事件，执行更新订单状态、计算盈亏、触发风控等逻辑

整个过程是异步的。事件源不用等处理器执行完，直接返回。这样系统就不会被慢操作拖死。

我的经验：刚开始做的时候，别搞太复杂。先实现一个「单队列 + 单线程」的版本，跑通了再上多线程、多队列。一步到位往往死得很难看。

7. 避坑指南

最后，分享几个我踩过的坑：

事件丢失：队列满了直接丢弃事件，但没有任何告警。后来我加了一个「丢弃计数器」，超过阈值就发报警。
死锁：处理器在处理事件时，又往总线发了一个事件，结果自己等自己……嗯，用异步队列可以解决。
性能瓶颈：所有事件走同一个队列，高优先级事件被低优先级事件堵住。后来改成多队列 + 优先级调度，问题解决。

好了，核心组件就这些。下一章我们聊聊「事件定义与序列化」——说白了就是事件长什么样，怎么在网络里传。到时候见。

📚 课程目录

核心组件设计：事件总线、事件处理器、事件队列、事件路由

1. 事件总线（Event Bus）

2. 事件处理器（Event Handler）

3. 事件队列（Event Queue）

4. 事件路由（Event Router）

5. 整体架构图

6. 组件协作流程

7. 避坑指南