“rea”这个代号在我本地仓库里躺了差不多两年。它的全称是 Real-time Event Analyzer一个轻量级的实时事件分析与分发引擎。简单说它解决的问题很常见业务系统里模块之间需要解耦通信但又不想为这件事引入一套需要单独部署和运维的消息中间件。rea 选择的是另一种思路——作为一个能嵌入业务进程、没有外部服务依赖、支持按主题订阅和过滤事件的小型事件总线在内存里完成事件的注册、路由、派发和监控。这篇文章我会把 rea 的设计背景、架构拆分、核心模块实现、工程化落地以及我在线上踩过的坑都翻出来讲清楚内容会比较偏实战适合正在折腾事件驱动架构、或者打算自己写一套轻量级事件分发框架的开发者参考。1. 为什么做 rea一次线上故障逼出来的项目1.1 事故复盘事件处理的两种坏味道事情要从一次典型的业务故障说起。当时我负责维护一个交易类的后端服务用户下单之后系统需要串行去更新库存、发送站内通知、写审计日志、给用户加积分。一个请求的处理链路长不说后面每次新增一个营销模块都要往主流程里再塞一段同步调用。代码越加越长接口越来越慢最后在一次大促流量下把数据库连接池打满接口直接雪崩。复盘的时候问题很明显主流程和附加动作耦合太深属于典型的“知道要解耦但不知道从哪下手”的状态。大家第一反应是上消息队列但评估之后发现为了处理一批内部业务事件要引入一套集群、配置生产消费组、维护如果挂了之后的监控告警对当时几十个节点的规模来说过于笨重。这就是事件处理的第一个坏味道不分场景一上来就上重型的独立中间件。第二种坏味道更隐蔽。有人提了一种“轻量方案”直接用一个全局的线程池每个模块往里丢任务。一开始看着清爽但接着问题就来了——没有任何订阅模型A模块想知道B模块发生了什么只能去硬编码调用想对某类事件单独过滤、加优先级、控制并发基本靠写if-else想追踪一条业务链路里各步骤的耗时得逐个去打点拼日志。线程池解决了异步却把路由和订阅的逻辑留在了业务代码里耦合只是换了个样子没有真正消失。1.2 rea 的定位与目标那次故障之后我开始认真思考一个大家都需要、但一直没被满足的中间档方案它得能嵌进应用进程不要求额外部署它需要有像消息队列那样的主题订阅能力让模块之间按事件类型解耦它还得有基本的控制能力比如优先级、重试、背压、可观测。这就是 rea 最初的轮廓。rea 明确不做的事情也很多。它不跨进程不持久化消息不追求极高的吞吐。它把自己定位成一个“库”而不是一个“系统”。这么做的好处很实际接入成本低一个依赖打包进去就能用没有网络故障没有磁盘占用心智负担小。代价是它有容量上限应用重启事件会丢这些在选型的时候就要接受。说清楚边界后面很多设计决策就顺理成章了。比如既然不跨进程就完全不用考虑序列化协议既然不持久化就可以放心用内存队列做缓冲既然是嵌入进程线程模型就必须充分尊重宿主应用的资源预算。rea 的核心目标可以归结成四句话嵌入式运行、按主题订阅、可动态匹配、可观测可治理。rea这个代号来源于我本地仓库里躺了差不多两年的项目全称是 Real-time Event Analyzer一个轻量级的实时事件分析与分发引擎。它的适用场景非常具体业务系统内部模块之间需要解耦通信但你又不想为这件事引入一套需要单独部署、维护、扩容的消息中间件。rea 的技术路线是做成一个能嵌入业务进程的事件总线在内存里完成事件的注册、路由、分发、重试和监控既保留消息主题订阅的灵活性又省去独立中间件的运维负担。这篇文章我会从设计背景、核心模块实现、工程化落地、线上问题排查几个维度完整复盘适合正在搞事件驱动架构、或者想自己写一套轻量级事件分发框架的开发者参考。全文基于我个人的实际开发经验代码方案属于常见实践的一种你可以按自己的场景改造。1. 背景与设计思路一个嵌入式事件分析引擎是怎么来的1.1 为什么不用现成的消息中间件先聊一个经常被问的问题市面上有那么多成熟的消息系统功能齐全、社区活跃为什么还要自己写一个rea答案不是别人做得不好而是场景不匹配。我手头这个业务很多时候只是想让模块A把某个状态变化告诉模块BB收到之后自己去做点事情。这种场景发生在同一个进程内部数据量不大实时性要求不高也不低最在意的是接入简单、别给我增加运维负担。用成熟消息中间件当然可以但代价也很清楚需要单独部署集群、需要配置生产消费组、需要处理网络分区甚至要考虑消息堆积时磁盘写满的问题。为了进程内的一个传递动作背上整套分布式系统的复杂度怎么算都不划算。我需要的是一种更轻的沟通方式像是办公室里两个人喊一嗓子而不是专门修一条物流专线。1.2 与观察者模式、响应式流的区别有人可能会说这不就是观察者模式吗Java 里监听器一加不就行了严格讲rea 确实脱胎于观察者模式但它解决的是观察者模式在复杂应用中不太够用的几个问题。第一个问题是路由粒度。观察者模式通常是一对一或者一对多直接通知而 rea 要做的是类似消息队列的主题路由。事件发布方不关心有哪些订阅者订阅者按主题表达兴趣事件总线负责匹配和分发。第二个问题是异步模型。纯观察者模式是同步调用的发布一个事件会阻塞到所有监听器执行完rea 则要把发布动作和订阅者的执行动作解耦采用异步派发。第三个问题是治理能力。观察者模式没有重试、优先级、背压、指标监控这些概念而这些在实际生产环境里恰恰是刚需。1.3 选型时的核心取舍再往后我还对比过响应式流。响应式流设计得很优雅背压机制尤其出色但它的编程模型对业务代码侵入性比较强很多 CRUD 团队的学习成本不低。rea 更想做成一种中间地带的轻量方案它保留事件驱动的好但用常见的命令式写法把 Buffer 和 Subscription 这些概念藏到引擎内部让业务代码只关心发一个事件和收一个事件。这个取舍很重要。很多技术方案失败不是因为性能差而是因为心智负担过高。rea 的定位从一开始就没有变过它应该像一个积累过经验的内部工具而不是一个需要团队成员重新学习的技术产品。2. 核心架构拆解事件信封、订阅注册表与分发器2.1 事件模型为什么设计成信封rea 的事件载体我称之为 EventEnvelope也就是事件信封。你可以把它理解成一个标准快递盒子盒子上贴了面单元数据盒子里装着真正要送的东西业务数据。实际实现的时候信封里包含六块信息字段类型作用eventIdString全局唯一事件ID用于链路追踪与去重topicString事件主题路由依据timestamplong事件产生时间用于延迟统计priorityint优先级决定派发顺序payloadMap业务数据载体traceIdString业务链路ID串联多个事件一个典型的订单事件长这样EventEnvelope envelope EventEnvelope.builder() .eventId(UUID.randomUUID().toString()) .topic(order.created) .timestamp(System.currentTimeMillis()) .priority(5) .traceId(traceId) .payload(orderDto) .build();注意 payload 用 Map 而不是强类型对象。这个设计是我刻意做的原因有两个一是在总线层面引擎只负责路由和投递不应该强依赖业务类型。如果用强类型对象订阅端就得做类型强制转换而且发布端一旦改了类结构所有订阅方都要跟着编译耦合就变强了。二是方便做内容过滤。有些订阅者只关心特定状态的订单比如已支付且金额大于1000元的订单这种条件在信封层面做预过滤比把整个业务对象暴露给匹配引擎要安全得多。信封设计本质上是在通用性和类型安全之间选了一个偏通用性的位置。2.2 订阅注册表用 Trie 树做主题匹配主题匹配是事件总线的心脏。最简单的方式是把所有订阅者放在一个 ConcurrentHashMap 里key 是完整主题字符串发布时用 event.topic 直接查。这个方案能解决精确匹配但解决不了通配符订阅。比如有个订阅者对所有订单相关事件都感兴趣写成 order.*。order.created、order.paid、order.cancelled 都该命中。如果只用哈希表即使能查到 order.created 和 order.paid也没法在 O(1) 时间内返回所有匹配结果。我的实现里改用了一棵站内自研的 Trie 树每个节点按点号切分主题段。class TopicTrie { private final MapString, TopicTrie children new ConcurrentHashMap(); private final SetSubscription subscriptions ConcurrentHashMap.newKeySet(); }插入状态下order.created 就是 order - created 两条边边尾节点挂上订阅者 Set。查询时如果事件主题是 order.paid就从根节点出发沿着 order 找到节点然后把该节点挂载的所有订阅者收集起来如果是精确匹配就只取最后一个 created 节点的订阅者如果主题段里出现了 *则做广度优先遍历把所有子节点的订阅者全部收集。这个结构的好处是插入和查询的时间复杂度只跟主题段数有关和订阅者总量无关。实际跑下来10 万层级订阅、每秒查询几千次耗时可忽略不计。2.3 分派线程模型派发线程与业务线程分离分派器是另一个需要仔细设计的模块。最容易犯的错误是让发布线程直接跑订阅者的回调。如果某个订阅者做了一个慢查询整个发布线程都会被拖住后面的所有事件跟着排队导致一个慢消费者拖垮整个发布链路。rea 把派发拆成了两条路径第一层是派发线程。发布方调用 publish 方法时事件不会直接回调给订阅者而是被投递到一个内部缓冲区与此同时得到发布确认。这个确认只代表事件已进入分派队列不代表所有订阅者都处理完了。单看这一个动作代价非常小发布方可以极速返回。第二层是业务线程。派发线程拿到订阅者列表之后把每个订阅者的执行任务提交到可配置的线程池。这一步是关键每个订阅者默认共用同一个全局线程池也可以单独配自己的线程池。这样做能让慢订阅者只会占用自己的配额不会阻塞全局派发。线程模型总结成一句话事件发布路径上的任何环节都不能被业务代码卡住。一旦出现长任务就必须丢出派发核心路径让它去抢业务线程的资源。2.4 重试与死信处理事件派发不总是一次成功的。订阅者抛异常或者执行超时都算失败。rea 需要一套重试机制但不能无限重试否则失败任务会积压拖垮系统。实现上比较朴素做了两层。第一层是同步重试单次派发时订阅者抛异常会立刻重试最多 N 次默认 3 次每次重试之间间隔指数退避比如 100ms、200ms、400ms。这个策略适合处理偶发的事务冲突或临时资源不足。第二层是死信队列。当订阅者重试 N 次仍然失败事件会进入一个容量有上限的环形缓冲区也就是死信队列。死信队列不会自动清理而是等着一个后台任务定期扫描把里面的事件重新投递一次或者输出一条带完整堆栈的错误日志。我当时额外加了一个订阅者级熔断逻辑某个订阅者在过去一分钟内连续失败超过 20 次引擎会把它暂时从派发列表里摘除进入一个冷却期。冷却期间所有发往该订阅者的事件都进入死信队列直到冷却结束再恢复。这个设计有效防止了一个濒临崩溃的模块反复拖累发布端。3. 核心实现细节与处理原则3.1 publish 的完整调用链把 rea 的核心 publish 流程完整走一遍这一段是很多初次阅读源码的人最关心的地方。public void publish(EventEnvelope event) { if (event null || event.getTopic() null) { return; } // 1. 根据优先级入队 priorityQueue.offer(event); // 2. 派发线程轮询取出事件 dispatcherThreadPool.execute(() - dispatchEnvelope(event)); } private void dispatchEnvelope(EventEnvelope event) { // 3. 订阅匹配 ListSubscription matched topicTrie.match(event.getTopic()); if (matched.isEmpty()) { metrics.recordDropped(event); return; } // 4. 按优先级排序 matched.sort(Comparator.comparingInt(Subscription::getOrder)); for (Subscription sub : matched) { try { // 5. 提交到订阅者线程池 sub.getExecutor().submit(() - safeInvoke(sub, event)); metrics.recordAccepted(sub, event); } catch (RejectedExecutionException ex) { // 6. 队列已满走死信 deadLetterQueue.offer(event); metrics.recordRejected(sub, event); } } } private void safeInvoke(Subscription sub, EventEnvelope event) { try { sub.getHandler().onEvent(event); metrics.recordSuccess(sub, event); } catch (Throwable t) { // 7. 失败重试 boolean ok sub.retry(event); if (!ok) { deadLetterQueue.offer(event); metrics.recordFailure(sub, event); } } }这段代码里最值得注意的点是 catch RejectedExecutionException。它对应的是背压场景某个订阅者的执行线程池已经满载无法再接任务。此时如果直接把事件丢弃会丢业务数据所以放到死信队列。如果死信队列也满了rea 的策略是当前线程同步阻塞直到死信队列腾出空间。这是唯一允许发布线程被阻塞的例外情况因为它代表整个系统资源已经见底再快也没有意义倒不如退一步防止雪崩。3.2 订阅者注册与注销注意内存泄漏订阅者的生命周期管理看着简单但非常容易在生产环境踩坑。最简单的方式是public Subscription subscribe(String topic, EventHandler handler) { Subscription subscription new Subscription(UUID.randomUUID().toString(), topic, handler); topicTrie.insert(topic, subscription); return subscription; }注册没问题问题在于反注册。如果一个监听器订阅了某个主题但持有这个订阅的对象被销毁时没有调用 unsubscribe那么订阅者会一直留在 Trie 树里。如果这个场景发生在动态创建对象的 Web 应用里等于每次请求都往 Trie 树里塞一个永久订阅内存会以肉眼可见的速度涨。rea 的解决方法很简单在 Subscription 里维护一个 AtomicBoolean active派发时先检查 active 状态如果已注销就直接跳过。同时提供一条最佳实践尽量使用项目自带的 IoC 容器管理订阅生命周期容器的 Bean 销毁时自动执行 unsubscribe。3.3 过滤器的实现方式过滤是事件总线的常见需求。rea 支持两种过滤器一种是预过滤在匹配订阅者时就进行条件判断不匹配的直接不投递另一种是后置过滤在订阅者回调里判断不满足条件的直接 return。在实际实现里预过滤用的是一组 Predicate 链Subscription sub eventBus.subscribe(order.created, handler) .filter(event - { OrderDto dto (OrderDto) event.getPayload(); return dto.getAmount() 1000; });底层其实是一个 ListPredicate 派发前逐个执行。我在这个模块上吃过一次亏把业务逻辑写进过滤器里导致过滤器自身抛异常会把整个订阅者都给弄死而且很难排查。后来规定过滤器里只允许做无副作用的只读判断不能发起网络请求或修改共享状态。这个原则后来救过团队很多次。4. 生产环境的工程化经验与性能数据4.1 分派器线程池怎么配线程池的大小是 rea 里最需要调参的地方。太大会造成大量线程切换太小又会让事件堆积。这不是一个拍脑袋的数字要结合订阅者的特点来算。我先给出一个常见生产环境的测试数据硬件是 8 核 CPUJava 17Linux 内核默认调度订阅主题数量订阅者数量全局线程池大小发布平均延迟P99 延迟每秒处理事件数1000500040.8ms8ms约 15 万1000500080.5ms5ms约 22 万10005000160.6ms6ms约 20 万你可能会奇怪从 8 到 16延迟反而恶化了。原因很简单8 核的机器16 个线程超过 CPU 核心数之后上下文切换开销开始抵扣并行收益。综合来看核心数等于线程池大小比较合理但如果订阅者执行体里有大量 IO 等待可以适当放大到两倍核心数。网络上的通用经验是CPU 密集 N1、IO 密集 2N我在 rea 里验证下来这个公式基本靠谱但最终值还得靠压测落定。4.2 事件延迟的观测手段rea 的发布延迟可以拆成三段入队耗时、匹配耗时、执行耗时。在存粹的进程内总线里入队和匹配都是微秒级的大头通常在执行。为了看到这三段数据我在 EventEnvelope 里埋了时间戳派发器在处理时用 System.nanoTime() 做记录long start System.nanoTime(); envelope.setEnqueueNanos(start); // 派发完成 envelope.setDispatchedNanos(System.nanoTime() - start); // 订阅者执行完成 envelope.setExecutedNanos(System.nanoTime() - start);最终这些数据会汇入一个 RingBuffer 指标流周期性地输出到日志或者监控系统。我强烈建议事件总线这种基础设施一定要从一开始就做指标埋点后期补会非常痛苦。此前见到的很多同行都因为前期没有埋点线上出了问题只能靠猜。4.3 配置项宁可少不要多rea 的配置项我刻意控制得很少只保留了五类全局线程池大小单订阅者最大线程池大小重试次数、重试退避基准时间死信队列容量熔断阈值、冷却时间这类基础设施最怕变成配置地狱。每多一个可选项就多一个线上事故的种子。宁可一开始写死等业务真的需要再放开也不要在第一个版本给用户二三十个旋钮。这个原则我在 rea 的多次迭代中受益良多。5. 常见问题与排查技巧实录5.1 订阅者执行慢事件堆积越来越严重现象发布端延迟没变但某个订阅者的队列深度持续上涨内存占用上升。排查第一件事是看线程池的活跃线程数和队列等待数。如果活跃线程数长期等于最大值说明业务执行确实饱和。接下来需要区分是CPU 被烧满还是线程被 IO 阻塞。如果是 CPU 烧满可以考虑为该订阅者单独分配线程池并限制事件并发度。如果是 IO 阻塞比如调远程 HTTP 服务重点要排查下游是否有超时设置。我在实际项目里遇到过的是某个订阅者会在回调里同步调用一个第三方接口第三方接口偶发 2 秒超时线程池很快被阻塞实例占满事件堆积到几十万。处理方案很直接把所有远程调用改为异步方式回调里只负责把数据放到一个待发送队列由单独线程慢慢消费。事件总线的回调函数应该是快进快出的任何可能阻塞超过几十毫秒的操作都不该放在里面。5.2 事件顺序错乱业务里有一些场景要求同一订单的事件严格有序比如 order.created 必须先于 order.paid 被处理。默认情况下 rea 使用全局线程池多个 worker 并发消费事件顺序必然被打乱。解决办法是给同一业务主键分配固定线程。我实现了一个一致性哈希路由取 event.getKey() 的哈希值对某个固定并发的 worker 数取模同一个 key 进入同一个 worker 串行执行。这样既保留了并发能力又保证了同一个业务实体的事件严格有序。这个方案的代价是 worker 数固定所有 key 分到这个池子上共享资源。实测下来如果 worker 数是 8事件量是 10 万/秒基本能满足 99% 的场景。只有那些单 key 事件每秒超过几万条的场景才需要考虑把大 key 单独拆出来。5.3 重复订阅导致重复消费场景同一个 Handler 被注册了两次事件发布后回调被执行两次而且能明显看到重复的副作用比如发了两次通知。排查路径看注册代码是不是被放在了某个循环里或者容器重启后没有清除上一次的订阅。持续追踪时发现一个有意思的情况——反射机制在重新加载类的时候会重建整个 Handler 对象但旧的 Trie 树节点没有清理干净于是新旧两份订阅同时存活。解决方案是统一管理订阅 ID每次订阅生成一个唯一 hash创建时记录在全局 registry 里。健康检查时如果发现同 topic、同 handler 类型的订阅超过一个就告警并把后注册的踢掉。另外从架构上建议订阅注册尽量收拢到少数几个配置类里别散落在业务代码各处。5.4 进程重启内存队列中的事件全部丢失这是我最开始没有解决的问题后来想通了。rea 明确不持久化所以进程崩溃、重启队列里未处理的事件就是会丢。这是设计选择不是 bug。如果你确实不能接受丢事件我建议不要把 rea 当成唯一的可靠性通道。做法是事件在发布时同步写一份到本地 WAL 日志文件重启后由 reaper 进程扫描日志文件重新投递。代价是每个事件多一次磁盘写吞吐会明显下降。我实际权衡下来绝大多数内部事件丢掉是可以容忍的真正不能容忍的事件本来就应该走外部可靠通道。5.5 排查工具清单我给 rea 配了一个简单的命令行诊断工具通过 JMX 暴露当前线程池状态、队列深度、死信数量、订阅者数量。每次线上出问题第一步都是先抓这几项快照再结合日志定位。如果你也为自己的中间件做类似的东西一定要让输出足够傻子友好别用二进制协议直接打可读文本。6. 后续扩展与个人经验总结rea 做到了第二个大版本的时候我开始觉得它的边界越来越清晰。后续可能的扩展方向有几个一个是做跨进程版本用本地 TCP 或者 UDP 在多个应用节点之间共享事件流此时才真正需要考虑序列化和网络分区的语义一个是做更细的流控比如按订阅者配额限制每秒最大派发量还有一个是集成更成熟的链路追踪系统让每个事件都能串联到业务全链路。如果你也想搞一个类似的轻量级事件引擎我建议先把最核心的事件模型、路由匹配、异步派发做好不要一上来就想把分布式和持久化做进去。很多问题其实是业务使用方式和线程模型没想清楚造成的跟技术栈关系不大。最后分享一个我踩过最深的坑给订阅者线程池设置队列容量时设得太小会导致大量事件进死信设得太大又会在流量高峰时拖垮内存。合理的经验是让队列容量等于线程数的 50 倍左右太大并不会带来性能提升只会延迟问题的暴露时间。真正稳妥的做法是配合监控做动态调整而不是迷信一个静态经验值。这些数字记在纸上不够得记在你的压测报告里。