第4章事件总线核心

4、事件总线核心：发布-订阅模式实现、事件通道设计、同步与异步分发机制

好，咱们今天聊点硬核的。

事件总线，说白了就是整个交易系统的「消息高速公路」。我见过不少团队，策略写得挺漂亮，但一到系统集成就崩——原因无他，事件总线没设计好。你想想看，行情来了，策略要响应，风控要检查，订单要发送，日志要记录……这些模块之间怎么通信？靠函数调用？那耦合度直接爆炸。

所以，我们需要一个事件总线。它负责把「谁关心什么事件」这件事管起来，然后高效地分发。

4.1 发布-订阅模式：解耦的基石

发布-订阅模式，其实不复杂。核心就三个角色：

发布者（Publisher）：只管把事件扔出来，不关心谁接。
订阅者（Subscriber）：只说自己对哪类事件感兴趣，不关心谁发。
事件通道（Event Channel）：中间人，负责匹配和转发。

我在项目中遇到过一种情况：某团队把行情处理直接写死在策略里。后来要加一个回测模块，得改策略代码。这就是典型的耦合灾难。用了事件总线之后，行情模块只管发 TickEvent，策略模块只管订阅 TickEvent，回测模块也订阅 TickEvent——大家各干各的，互不干扰。

核心原则：发布者和订阅者之间，绝对不要有直接引用。它们只通过事件类型和事件数据来沟通。

来看一个最简实现：

// 事件基类
public abstract class Event
{
    public DateTime Timestamp { get; set; }
}

// 具体事件
public class TickEvent : Event
{
    public string Symbol { get; set; }
    public double Price { get; set; }
    public double Volume { get; set; }
}

// 事件总线（简化版）
public class EventBus
{
    private Dictionary<Type, List<Action<Event>>> _subscribers = new();

    public void Subscribe<T>(Action<T> handler) where T : Event
    {
        var eventType = typeof(T);
        if (!_subscribers.ContainsKey(eventType))
            _subscribers[eventType] = new List<Action<Event>>();

        _subscribers[eventType].Add(e => handler((T)e));
    }

    public void Publish<T>(T eventData) where T : Event
    {
        var eventType = typeof(T);
        if (!_subscribers.ContainsKey(eventType)) return;

        foreach (var handler in _subscribers[eventType])
        {
            handler(eventData);
        }
    }
}

嗯，这里要注意：上面的代码是同步的。实际生产环境，我们得考虑更多。

4.2 事件通道设计：别把所有鸡蛋放一个篮子里

我建议把事件通道按「优先级」和「业务域」来划分。为什么？

你想想看，行情事件每秒可能上千次，而用户登录事件一天也没几次。如果它们走同一个通道，高频率事件会阻塞低频率事件的处理。我曾经在一个项目中吃过这个亏——行情处理线程把事件队列塞满了，导致用户下单的确认事件延迟了200毫秒。200毫秒在交易里，可能就是一笔滑点损失。

所以，我一般这样设计通道：

通道名称	优先级	典型事件	处理要求
MarketDataChannel	最高	TickEvent, OrderBookEvent	低延迟，可丢部分非关键数据
OrderChannel	高	OrderSubmitted, OrderFilled	必须可靠，顺序保证
RiskChannel	高	RiskCheckPassed, RiskViolation	必须可靠，实时性要求高
LogChannel	低	LogEvent, MetricsEvent	可以批量，允许一定延迟
SystemChannel	中	ConfigChanged, Heartbeat	可靠，但不要求实时

我的习惯：每个通道独立一个线程/协程来处理。MarketDataChannel 用环形缓冲区（RingBuffer）来减少GC压力。OrderChannel 用阻塞队列保证顺序。LogChannel 用无界队列，后台慢慢刷。

4.3 同步与异步分发：选对路子，事半功倍

同步还是异步？这不是一个二选一的问题，而是「在什么地方用什么」的问题。

4.3.1 同步分发

同步分发，就是发布者调用 Publish 之后，要等所有订阅者处理完才返回。好处是简单、可控、顺序保证。坏处是——慢。如果某个订阅者处理慢了，会拖累整个发布流程。

我一般在以下场景用同步：

风控检查：订单发出去之前，必须等风控模块返回结果。这时候异步反而危险。
状态更新：比如账户余额变更，需要立即生效。
本地事件链：事件A的处理结果会触发事件B，且B必须在A之后立即执行。

// 同步分发示例
public void PublishSync<T>(T eventData) where T : Event
{
    var eventType = typeof(T);
    if (!_subscribers.ContainsKey(eventType)) return;

    foreach (var handler in _subscribers[eventType])
    {
        try
        {
            handler(eventData);  // 同步调用，等待返回
        }
        catch (Exception ex)
        {
            // 记录异常，但不要中断其他订阅者
            _logger.Error($"Handler failed for {eventType.Name}: {ex.Message}");
        }
    }
}

4.3.2 异步分发

异步分发，就是发布者把事件扔到队列里，然后立刻返回。订阅者在另一个线程里慢慢处理。好处是不阻塞发布者，适合高吞吐场景。坏处是——顺序可能乱，而且处理结果不能直接返回给发布者。

我建议在以下场景用异步：

行情处理：每秒几千次Tick，同步处理会卡死。
日志记录：不需要实时反馈。
指标计算：比如计算夏普比率，晚几秒没关系。

// 异步分发示例（使用BlockingCollection）
public class AsyncEventBus
{
    private BlockingCollection<Event> _queue = new();
    private CancellationTokenSource _cts = new();

    public AsyncEventBus()
    {
        // 启动后台处理线程
        Task.Run(() => ProcessQueue(_cts.Token));
    }

    public void PublishAsync<T>(T eventData) where T : Event
    {
        _queue.Add(eventData);  // 立刻返回
    }

    private void ProcessQueue(CancellationToken token)
    {
        foreach (var evt in _queue.GetConsumingEnumerable(token))
        {
            // 查找并调用订阅者
            Dispatch(evt);
        }
    }
}

我曾经踩过的坑：异步分发时，如果订阅者抛出异常，整个消费线程会挂掉。一定要在 Dispatch 方法里加 try-catch，并且记录异常后继续处理下一个事件。否则，一个坏掉的订阅者会拖垮整个系统。

4.4 混合模式：同步+异步的实战方案

实际项目中，我很少只用一种模式。更常见的做法是：

发布者调用 Publish 时，传入一个「分发模式」参数。
事件总线根据事件类型或订阅者的标记，决定同步还是异步。
对于需要返回结果的场景（比如风控），用同步+回调的方式。

举个例子：

public enum DispatchMode
{
    Sync,
    Async
}

public class HybridEventBus
{
    private Dictionary<Type, List<SubscriberEntry>> _subscribers = new();
    private AsyncEventBus _asyncBus = new();  // 内部异步总线

    public void Subscribe<T>(Action<T> handler, DispatchMode mode = DispatchMode.Sync)
    {
        var entry = new SubscriberEntry
        {
            Handler = e => handler((T)e),
            Mode = mode
        };
        // ... 注册逻辑
    }

    public void Publish<T>(T eventData) where T : Event
    {
        var eventType = typeof(T);
        if (!_subscribers.ContainsKey(eventType)) return;

        foreach (var entry in _subscribers[eventType])
        {
            if (entry.Mode == DispatchMode.Sync)
            {
                entry.Handler(eventData);  // 同步
            }
            else
            {
                _asyncBus.PublishAsync(eventData);  // 异步
            }
        }
    }
}

我的建议：默认用同步，只有明确知道某个事件处理耗时较长且不需要立即反馈时，才用异步。别一上来就全异步，调试起来会让人崩溃。

4.5 事件总线架构图

下面这张图，是我个人习惯的事件总线内部结构。你可以看到，事件从发布者出来，经过通道路由，再根据分发模式走不同的路径。

从图上你能看到：同步路径直接调用订阅者，异步路径则先入队列，再由后台线程消费。这样设计，既保证了关键事件的实时性，又让非关键事件不拖累主流程。

4.6 避坑指南

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

事件顺序问题：异步分发时，如果两个事件有因果关系（比如先下单再成交），必须保证它们被同一个线程处理。我一般用「事件分区」——按订单ID哈希，确保同一个订单的事件走同一个线程。
内存泄漏：订阅者忘记取消订阅，会导致事件总线持有对它的引用，GC无法回收。我习惯在订阅时返回一个「取消令牌」，让调用方在不再需要时主动取消。
死锁：同步分发时，如果订阅者又反过来发布事件，可能形成循环调用。我加了一个「重入检测」——如果检测到当前线程已经在处理同一事件类型，就抛异常或异步化。
性能瓶颈：事件总线的锁竞争是个大问题。我后来改用「无锁队列」+「每个通道独立锁」的方式，把性能提升了3倍。

一句话总结：事件总线是交易系统的「神经系统」。设计得好，模块之间松耦合、高内聚；设计得不好，系统就像一盘散沙，改一处动全身。

好了，这一章的内容就到这里。事件总线的核心你已经掌握了——发布-订阅模式、通道设计、同步异步分发。下一章，我们会把这些知识落地到实际代码中，搭建一个真正能跑的事件驱动框架。

📚 课程目录

4、事件总线核心：发布-订阅模式实现、事件通道设计、同步与异步分发机制

4.1 发布-订阅模式：解耦的基石

4.2 事件通道设计：别把所有鸡蛋放一个篮子里

4.3 同步与异步分发：选对路子，事半功倍

4.3.1 同步分发

4.3.2 异步分发

4.4 混合模式：同步+异步的实战方案

4.5 事件总线架构图

4.6 避坑指南