第4章事件总线测试

4. 事件总线测试：事件发布/订阅机制验证、事件优先级队列测试、事件超时与重试机制

事件总线，说白了就是整个交易系统的「神经中枢」。

我参与的第一个量化系统，就是因为事件总线没测好，导致行情来了订单发不出去。那次事故让我明白——总线崩了，系统就瘫了。所以这一章，咱们重点啃三块硬骨头：发布/订阅机制、优先级队列、超时与重试。

4.1 事件发布/订阅机制验证

先问个问题：你怎么确认一个事件真的被「发布」出去了？

很多人只测「发没发」，不测「谁收到了、收了几次、顺序对不对」。我个人习惯用回调计数器 + 订阅者快照来做验证。

  核心验证点：
  同一个事件能否被多个订阅者接收
订阅者取消订阅后是否还能收到
通配符订阅（如订阅所有行情事件）是否生效
重复订阅同一事件是否导致重复消费

举个例子，我曾经遇到一个坑：某个策略订阅了「成交回报」事件，但代码里不小心注册了两次回调。结果每次成交，策略执行了两次——仓位算错了，直接爆仓。嗯，从那以后我强制要求：订阅必须去重。

// 伪代码：验证发布/订阅
EventBus bus = new EventBus();

int count1 = 0, count2 = 0;
bus.subscribe("trade.fill", () -> count1++);
bus.subscribe("trade.fill", () -> count2++);

bus.publish("trade.fill", new FillEvent("BTC", 1.0));

assert count1 == 1 : "订阅者1未收到";
assert count2 == 1 : "订阅者2未收到";

// 取消订阅后验证
bus.unsubscribe("trade.fill", subscriber1);
bus.publish("trade.fill", new FillEvent("ETH", 2.0));
assert count1 == 1 : "取消订阅后仍收到事件";
assert count2 == 2 : "订阅者2应继续接收";

我的小技巧： 用原子计数器（AtomicInteger）做并发场景下的订阅验证。单线程测不出竞态问题，但交易系统往往是多线程的。

4.2 事件优先级队列测试

交易系统里，不是所有事件都「平等」的。

比如「风控强平指令」必须比「普通订单更新」先处理。如果优先级搞反了，后果很严重。我见过一个系统，因为把「行情刷新」事件优先级设得比「止损指令」还高，结果行情剧烈波动时，止损单被堵在队列后面，等处理时已经穿仓了。

优先级队列测试，我一般关注三点：

高优先级事件是否被优先消费——哪怕它后到
同优先级事件是否保持FIFO——先来后到不能乱
优先级反转是否被处理——低优先级事件会不会阻塞高优先级

优先级等级	事件类型	最大延迟要求
P0（最高）	风控强平、止损触发	< 1ms
P1	订单成交回报、撤单确认	< 10ms
P2	行情快照、账户变动	< 100ms
P3（最低）	日志、统计、非实时通知	可容忍秒级

// 优先级队列测试用例
PriorityQueue<Event> queue = new PriorityQueue<>(Comparator.comparingInt(Event::getPriority));

// 先塞一个P2事件，再塞一个P0事件
queue.add(new Event("行情刷新", 2));
queue.add(new Event("强平指令", 0));

Event first = queue.poll();
assert first.getType().equals("强平指令") : "高优先级事件应被优先消费";

// 同优先级测试：先来后到
queue.add(new Event("订单A", 1));
queue.add(new Event("订单B", 1));
Event a = queue.poll();
Event b = queue.poll();
assert a.getId() < b.getId() : "同优先级应保持FIFO";

注意： 别用无界队列！我曾经在生产环境遇到过内存溢出——因为低优先级事件堆积太多，把堆内存撑爆了。建议用有界优先级队列，并设置拒绝策略（比如丢弃最低优先级事件）。

4.3 事件超时与重试机制

事件处理不是总能成功的。网络闪断、数据库锁等待、下游服务挂了……这些情况在交易系统里太常见了。

我个人的原则是：该重试的重试，该超时的超时，别死等。

4.3.1 超时机制测试

超时测试的核心就一句话：事件处理超过阈值后，系统怎么反应？

是直接丢弃？还是放入死信队列？
是否触发告警？
超时后是否释放线程资源？

我记得有一次测试，发现超时后线程池里的线程数一直在涨——原来超时事件只是打了个日志，线程还在阻塞等待。这会导致线程池耗尽，后续事件全部排队。说白了，这就是个隐形炸弹。

// 超时测试：模拟处理耗时过长
EventBus bus = new EventBus(500); // 超时阈值500ms

bus.subscribe("slow.event", event -> {
    Thread.sleep(1000); // 模拟慢处理
});

long start = System.currentTimeMillis();
try {
    bus.publishSync("slow.event", data);
} catch (TimeoutException e) {
    long elapsed = System.currentTimeMillis() - start;
    assert elapsed < 600 : "超时检测延迟过高";
    // 验证事件进入死信队列
    assert bus.getDeadLetterQueue().size() == 1;
}

避坑指南： 我曾经把超时时间设得太短（50ms），结果正常处理也被超时了。建议超时阈值 = 正常处理P99延迟 × 3，留足余量。

4.3.2 重试机制测试

重试不是简单的「失败了再来一次」。你得考虑：

重试次数上限——无限重试等于死循环
重试间隔策略——固定间隔？指数退避？
幂等性——重试不会导致重复下单或重复记账
最终失败处理——重试耗尽后怎么办？

我见过最离谱的案例：某个订单提交失败后，系统每隔1秒重试一次，重试了整整3天——因为没人设置最大重试次数。最后数据库里多了几万条重复订单记录。

// 重试机制测试：指数退避 + 最大重试次数
RetryPolicy policy = RetryPolicy.builder()
    .maxRetries(3)
    .initialDelay(100)  // 首次重试延迟100ms
    .backoffMultiplier(2) // 每次翻倍：100ms -> 200ms -> 400ms
    .build();

EventProcessor processor = new EventProcessor(policy);

// 模拟前两次失败，第三次成功
AtomicInteger attempt = new AtomicInteger(0);
processor.process("order.submit", event -> {
    int n = attempt.incrementAndGet();
    if (n < 3) throw new RetryableException("模拟失败");
    return true;
});

assert attempt.get() == 3 : "应重试3次";
assert processor.getMetrics().getRetryCount() == 2;
assert processor.getMetrics().getSuccessCount() == 1;

// 测试重试耗尽
processor.process("order.submit", event -> {
    throw new RetryableException("永远失败");
});
assert processor.getMetrics().getFailedCount() == 1;
assert processor.getDeadLetterQueue().size() == 1;

  关键原则：
  重试只对「临时性失败」有效（网络超时、锁冲突）
对「永久性失败」（参数错误、余额不足）不要重试，直接报错
重试间隔必须带随机抖动（jitter），防止所有重试同时打爆下游

4.4 本章知识体系总览

下面这张图是我自己梳理的事件总线测试框架，你可以照着这个结构去设计你的测试用例：

说实话，事件总线测试是量化系统里最容易「看起来没问题，一上线就出事」的环节。我建议你把这章的测试用例做成自动化回归集，每次代码变更都跑一遍。别问我为什么——问就是吃过亏。

📚 课程目录

4. 事件总线测试：事件发布/订阅机制验证、事件优先级队列测试、事件超时与重试机制

4.1 事件发布/订阅机制验证

4.2 事件优先级队列测试

4.3 事件超时与重试机制

4.3.1 超时机制测试

4.3.2 重试机制测试

4.4 本章知识体系总览