2. 核心概念解析：事件、事件源、事件处理器、事件循环
好，咱们正式开始啃核心概念。
这四个词，说白了就是事件驱动引擎的「四根柱子」。你搞懂了它们，整个系统的骨架就清晰了。我当年刚接触这个架构时，也花了不少时间才理清它们之间的关系。今天咱们一个一个拆开讲。
2.1 事件（Event）—— 系统的「神经信号」
事件是什么？简单说，就是「发生了某件事」这个信息本身。
在交易系统里，事件可以是一个订单被成交了，可以是一笔行情数据到了，也可以是风控规则被触发了。它不包含业务逻辑，只包含「发生了什么」以及「相关的数据」。
我个人习惯把事件设计成不可变对象。一旦创建，就不能修改。为什么？因为事件可能在多个处理器之间传递，如果谁都能改，你根本追查不到问题出在哪一步。我在一个早期项目里吃过这个亏，后来所有事件都强制用只读结构。
一个典型的事件结构长这样：
@dataclass(frozen=True)
class TradeEvent:
event_id: str
event_type: str # 比如 "ORDER_FILLED"
symbol: str
price: float
quantity: int
timestamp: float
extra: dict = field(default_factory=dict)注意几个关键点：
- event_type：用字符串常量，别用数字。可读性第一。
- timestamp：统一用浮点数或纳秒级整数。我建议用纳秒，高频场景下差几微秒都可能出问题。
- extra：兜底字段。你永远不知道未来会加什么信息，留个字典最灵活。
小技巧：事件类型建议用枚举或常量类管理。我曾经见过有人直接写字符串 "fill"、"FILL"、"filled" 混用，排查了一天。统一管理，省心。
2.2 事件源（Event Source）—— 谁在「生产」事件？
事件源就是事件的「生产者」。它负责监听外部或内部的变化，然后包装成事件丢进系统。
常见的几类事件源：
| 事件源类型 | 举例 | 说明 |
|---|---|---|
| 外部数据源 | 行情网关、交易所API | 接收网络数据，解析成事件 |
| 内部定时器 | 心跳事件、定时任务 | 按时间周期触发 |
| 用户操作 | 手动下单、参数调整 | GUI或CLI触发 |
| 系统内部 | 风控模块、订单管理 | 模块间通信产生的事件 |
事件源的核心职责只有一个：把原始信号转成标准事件。它不需要关心事件被谁处理、怎么处理。
class MarketDataSource:
def __init__(self, websocket_url: str):
self.ws_url = websocket_url
def start(self, event_bus):
# 伪代码：连接行情，收到数据就发布事件
while True:
raw_data = self.ws.receive()
event = self._parse(raw_data)
event_bus.publish(event) # 只负责发布，不负责处理注意：事件源不要做耗时操作。我曾经见过有人在行情解析里做复杂计算，结果阻塞了数据接收，导致行情延迟。事件源只做「接收→包装→发布」这三步，其他事情交给处理器。
2.3 事件处理器（Event Handler）—— 真正干活的「工人」
事件处理器就是处理事件的逻辑单元。它订阅特定类型的事件，然后执行对应的业务逻辑。
一个处理器只做一件事。这是原则。
举个例子：
class OrderFillHandler:
def __init__(self, portfolio_manager):
self.pm = portfolio_manager
def handle(self, event: TradeEvent):
if event.event_type != "ORDER_FILLED":
return # 不关心的直接跳过
# 更新持仓
self.pm.update_position(event.symbol, event.quantity, event.price)
# 记录日志
logger.info(f"成交: {event.symbol} {event.quantity}@{event.price}")
# 触发后续策略逻辑
self.pm.check_strategy_triggers(event.symbol)你想想看，如果我把成交处理、风控检查、日志记录、策略触发全塞到一个处理器里，会怎样？
- 代码耦合，改一处可能影响所有逻辑
- 难以测试，你得模拟各种场景
- 扩展困难，加一个新功能就得改老代码
我的经验：每个处理器只订阅它需要的事件类型。用装饰器或注册表来声明订阅关系，比硬编码 if-else 优雅得多。
2.4 事件循环（Event Loop）—— 系统的「心脏」
事件循环就是那个永不停歇的「调度器」。它从事件队列里取出事件，然后分发给对应的处理器。
说白了，就是一个 while True 循环，但里面的门道不少。
class EventLoop:
def __init__(self):
self.queue = asyncio.Queue()
self.handlers = {} # event_type -> [handler1, handler2, ...]
def register(self, event_type: str, handler):
self.handlers.setdefault(event_type, []).append(handler)
async def run(self):
while True:
event = await self.queue.get()
handlers = self.handlers.get(event.event_type, [])
for handler in handlers:
try:
await handler.handle(event)
except Exception as e:
logger.error(f"处理器异常: {e}", exc_info=True)
# 这里要决定：继续还是终止？事件循环要处理几个关键问题：
| 问题 | 解决方案 | 我的建议 |
|---|---|---|
| 处理器异常 | 捕获异常，记录日志，继续运行 | 别让一个处理器拖垮整个系统 |
| 事件积压 | 监控队列长度，触发背压机制 | 设置最大队列长度，超限就丢弃或降级 |
| 处理器耗时 | 异步处理，或使用线程池 | 高频场景下，处理器必须毫秒级完成 |
| 事件顺序 | 同一标的的事件保持FIFO | 用分区队列，按symbol分桶 |
避坑指南：我曾经在一个项目里没做异常隔离，结果一个处理器抛异常后，整个事件循环挂了。所有订单都卡住了，那叫一个惨。记住：事件循环必须健壮到能处理任何异常。
2.5 四者关系：一张图说清楚
这四个概念的关系，我用一张 SVG 图来展示：
这张图你看懂了吗？数据流是单向的：事件源产生事件 → 放入队列 → 事件循环取出 → 分发给处理器。没有回头路，没有交叉依赖。
核心原则：事件源不知道处理器存在，处理器不知道事件源是谁，事件循环只负责调度不负责业务。各司其职，这就是事件驱动架构的精髓。
嗯，这四个概念就讲到这里。你可能会觉得「就这？」，但相信我，真正落地时，每个环节都有坑等着你。后面几章咱们会深入每个模块的实战细节。
公众号:蓝海资料掘金营，微信deep3321