如果你所在的业务系统慢慢变成了一张拆不开的蜘蛛网模块之间互相创建互相调用时间一长你大概率会认真考虑一件事让状态变化自己长腿跑起来。我最近在内部项目里把之前随手写的一个小模块正式抽成了独立组件代号就叫 REAResource Event Agent资源事件代理。它解决的问题说大不大、说小不小多个数据源同时变化的时候怎么把变化事件稳定、有序、不重复地送到所有关心它的地方。这篇文章是我从设计、编码、踩坑到压测的完整记录也适合想自己搭一套轻量事件机制的开发者参考。1. REA 这个项目是怎么冒出来的1.1 事件驱动架构里的一个尴尬空白我所在的业务线维护着一个中台系统订单、库存、积分、消息通知四块业务强耦合。最早的代码是典型的“调用链式”写法订单状态一变控制器直接去调库存服务库存服务扣完又要通知积分服务积分加完还要触发消息推送。每加一个联动环节就要改一圈上游调用方测试起来也异常痛苦经常出现“一个库存接口改了订单模块莫名挂掉”的连锁反应。后来我们尝试引入事件驱动思维某个数据源变了发一个事件谁关心谁自己订阅。这个思路本身没有错但落地的时候发现市面上能用的方案处于两个极端。一头是重量级的消息中间件部署、运维、分区、消费组全都要管只是为了几个进程内的模块联动实在有点大炮打蚊子。另一头是轻量的发布订阅库这类库往往只做了一件事on和emit。问题在于它们没有“资源快照”和“版本变化”的概念事件发得倒是快但订阅方根本不知道该不该处理也没有办法判断当前自己拿到的状态是不是最新状态。那个尴尬的空白刚好卡在中间我需要一个既轻量又带一点业务感知的事件骨架。它能理解“某个资源发生了变化”能根据资源的版本号做去重还能把事件按照固定的顺序送给多个处理方。于是 REA 就诞生了。它的定位从一开始就不是通用的消息中间件而是进程内的数据资源与业务事件之间的翻译层。1.2 为什么自己写而不直接用现成方案决定自研之前我大概花了一个晚上把现有方案做了个选型对比。当时考虑过三条路通用事件总线库、全局状态管理框架、消息队列中间件。它们的核心差异其实非常明显。方案方向优点不足适配场景通用事件总线库轻量、API 简单、上手快没有资源概念与事件去重订阅方容易重复处理简单的 UI 通知、组件解耦全局状态管理框架带响应式状态、组件联动方便对自定义业务事件不够友好模板化严重前端页面状态同步消息队列中间件可靠、可扩展、跨服务部署和运维成本高进程内场景太重分布式服务间事件流转自研 REA贴合自家业务模型能控制细节需要自己维护中台内部的资源联动我们真正需要的组合是以资源为核心以事件为信号以代理为处理方。资源负责提供当前快照和版本号事件负责描述“什么变了”代理负责响应变化。这三者缺一不可。通用事件总线库能给我事件和代理但给不了资源快照全局状态管理框架能给我资源和联动但事件表达又被捆得很死。与其在别人的抽象上层打补丁不如写一个只有八百行核心代码的小模块按我们的业务习惯来建模。事实证明这个决定是对的。项目上线后新业务接入 REA 基本只需要定义一个新的资源类型和几个代理不需要修改核心逻辑。而且因为代码在自己手里后续加错误隔离、幂等控制、顺序屏障等等都变得非常直接不要求着某个开源社区改需求。2. REA 的核心模型资源、事件、代理三者怎么配合2.1 Resource 抽象把数据源统一成一个接口REA 里最核心的概念是资源Resource。任何能被观察的数据源比如一个订单记录、一个库存数字、一个用户积分账户都可以抽象成 Resource。它不需要知道自己的业务语义只需要提供三个能力标识、快照、版本号。export interface ResourceTState unknown { readonly id: string; readonly version: number; getSnapshot(): TState; }version是整数每次资源状态变化时自增。REA 在派发事件之前会比较版本号如果新事件携带的版本和当前注册表里的版本相同就说明这次变化已经被处理过直接丢弃。这样设计避免了在派发路径上做深比较省掉了大量无谓的序列化和对象比对开销。getSnapshot()的作用是让代理在处理事件时能拿到资源的当前状态。很多事件总线只传事件内容不提供“当前值”导致代理必须自己维护一份状态缓存既容易过期也容易不一致。REA 把资源快照挂在注册表上代理按需读取永远取到最新值。实际建模的时候我不会让业务对象直接实现这个接口而是用一个适配器包一层。比如订单服务里有自己的订单实体我就写一个OrderResourceAdapter把实体的变化映射成 REA 的资源版本更新。这样 REA 完全不侵入业务存储层。2.2 Event 定义避免字符串魔法早期原型里我用的是字符串事件名例如emit(order:updated, data)。第一天用着很爽第二天就发现问题事件名靠字符串硬编码搜索起来麻烦写错一个字母也不会报错而且没法从事件本身知道它是从哪个资源发出来的。后来我把事件定义成了结构化的DomainEventexport type EventType string | symbol; export interface DomainEventTMeta unknown { type: EventType; sourceResourceId: string; sequence: number; occurredAt: number; meta: TMeta; }每个事件都绑定了sourceResourceId这样代理可以按来源资源过滤。sequence是全局单调递增的序号用来判断事件的先后关系。occurredAt是发生时间方便排查问题时做时间线还原。meta才是真正的事件载荷。这里有一个很重要的设计决策事件类型允许symbol。字符串事件名虽然直观但容易在不同模块间产生碰撞用symbol则天然隔离。我实际项目里采用的是字符串常量加as const断言便于日志打印同时约定所有事件类型集中定义在一个文件里不允许随便散落。事件不是凭空产生的。REA 要求所有事件必须由一个资源变化触发这是刻意设计的约束。事件如果满天飞没有归属系统最终会退化成纯广播总线等于退回了老路。2.3 Agent 生命周期注册、派发、销毁有了资源和事件还需要一个处理方REA 里叫代理Agent。代理就是一个对象声明自己关心哪些事件类型并提供onEvent方法处理事件。export interface Agent { id: string; interests: EventType[]; onEvent(event: DomainEvent): void | Promisevoid; }注册代理的入口长这样function subscribe(agent: Agent): () void { // 在内部索引中登记 agent // 返回一个取消订阅函数 }这个返回的取消函数非常重要。我们项目里所有代理注册的代码都必须保存返回值并在合适的生命周期钩子里执行。注册是第一步注销同样不可以省略。代理销毁时不仅要取消订阅还要处理自己维护的临时状态。比如一个聚合统计代理内存里缓存了一批待处理事件注销前要把这些数据落库或清理否则就会静默丢失。REA 本身不负责代理内部状态但它提供了一个dispose()钩子的约定代理实现者需要自己保证清理干净。3. 从零实现 REA 的关键代码与设计取舍3.1 内核实现存储结构选择REA 的内核我写得很克制核心存储就是两层 Map。第一层resources根据资源 id 定位资源条目第二层agents根据代理 id 定位代理条目。资源条目内部维护了该资源触发的事件类型到代理集合的映射。interface ResourceEntry { resource: Resourceunknown; listeners: MapEventType, SetAgent; } class ReaRegistry { private resources new Mapstring, ResourceEntry(); private agents new Mapstring, Agent(); }为什么用SetAgent而不是数组因为代理可能因为业务逻辑重复注册数组去重需要indexOf遍历Set天然保证唯一而且删除订阅时平均开销更低。为什么要在资源条目里维护listeners因为派发时我是从资源出发找代理的这样可以避免扫描全部代理再做事件类型匹配查找复杂度是 O(1)。注册代理的时候REA 会遍历订阅者关心的所有事件类型把代理挂到对应资源的监听表里。这个遍历是注册时一次性完成的派发时的热路径不会出现类型匹配的循环。function subscribe(agent: Agent): () void { if (this.agents.has(agent.id)) { throw new Error(agent duplicated: ${agent.id}); } this.agents.set(agent.id, agent); for (const type of agent.interests) { for (const resource of this.resources.values()) { let listeners resource.listeners.get(type); if (!listeners) { listeners new Set(); resource.listeners.set(type, listeners); } listeners.add(agent); } } return () { this.agents.delete(agent.id); for (const resource of this.resources.values()) { for (const listeners of resource.listeners.values()) { listeners.delete(agent); } } }; }真实项目里我还会在这种遍历前加一层事件类型索引避免subscribe对全部资源做无意义扫描。不过对于几百个资源以内的系统上面的简单实现反而更容易读。3.2 异步事件派发用微任务还是同步循环这是开发中拿不准的部分之一。刚开始我采用的是同步派发资源写入方法里直接调用emitemit立刻遍历代理并执行onEvent。同步派发的优点是时序清晰出错了能在调用栈里直接定位到是哪一次资源写入引发的。但很快踩了坑。某个代理在处理完事件后会反过来更新同一个资源资源更新又触发新事件新事件又进入代理。一旦有循环依赖调用栈就会无限增长最后直接爆栈。就算没有循环同步派发也会让一个代理的错误中断整个资源写入流程把事件处理的异常和业务写入的异常搅在一起。于是我改成了“同步收集、异步派发”。资源写入后事件先进入 pending 队列然后通过微任务调度统一 flush。class ReaCore { private pending: DomainEvent[] []; private scheduling false; protected emit(event: DomainEvent) { this.pending.push(event); this.scheduleFlush(); } private scheduleFlush() { if (this.scheduling) return; this.scheduling true; queueMicrotask(() { this.flush(); }); } private flush() { this.scheduling false; const events this.pending; this.pending []; for (const event of events) { this.dispatchToInterestedAgents(event); } } }这里的关键是 pending 队列的折叠效应。同一轮同步代码里连续写入多个资源每个资源产生的多个事件会合并到同一次 flush 中处理。微任务会在当前宏任务同步执行完毕后统一启动因此写入逻辑可以安全返回代理处理时也不会打断业务操作。顺带纠正一个常见误区微任务和微任务之间不会存在事件循环插入所以同一批 pending 中的事件顺序是确定的。但跨资源事件的业务顺序不能只靠微任务来保证这一点后面踩坑部分细说。3.3 错误隔离与死循环防护事件代理最怕一个代理出错连累全局。我在flush循环里给每个代理的onEvent都单独包了一层 try/catch任何代理抛出的异常都不会中断同一批其他代理的执行。for (const agent of listeners) { try { const result agent.onEvent(event); if (result instanceof Promise) { result.catch((err) this.handleError(err, agent, event)); } } catch (err) { this.handleError(err, agent, event); } }同步异常和异步异常分开处理。同步异常立刻被 catch异步异常通过 Promise catch 兜住。handleError只做三件事记录日志、写入错误队列、抛出全局错误事件。全局错误事件本身也可以被代理订阅方便做告警。死循环防护用的是“同轮窗口去重 派发深度计数”。资源更新事件如果在极短时间内被重复派发REA 会记录sourceResourceId type version的组合同一个微任务窗口内相同组合只派发一次。另外每轮 flush 内对单个代理设置派发深度上限超过阈值就跳过后续派发并告警。这两个机制合在一起至少能把“订单完成更新订单状态又触发订单完成”这类循环扼杀在早期。4. 接入业务系统时最容易踩的坑4.1 订阅泄漏一个真实的内存上涨排查REA 第一次被集成进某个报表页面时QA 反馈页面反复打开关闭后内存持续上涨。我一开始怀疑是跨域接口的问题但打开浏览器的内存分析工具做了两次堆快照对比发现增长的主体非常集中一个业务代理对象出现了几百个实例数量和页面历史打开次数一一对应。问题不在 REA 核心而在调用方注册页面代理时只调了rea.subscribe()没有把返回的取消函数挂到页面卸载钩子里。页面关了以后代理对象依然留在注册表中引用链没断内存自然不被回收。修复很简单const cleanup: Array() void []; // 页面初始化 cleanup.push(rea.subscribe(orderAgent)); cleanup.push(rea.subscribe(stockAgent)); // 页面卸载时统一执行 cleanup.forEach((unsubscribe) unsubscribe());我把这条经验固化成了两个约定。第一业务方接入 REA 时必须保存并执行 unsubscribe 返回值。第二REA 在注册代理时如果检测到同 id 代理重复注册会直接抛出异常。这个约定一开始看起来有点严格后来救了好几次线上“幽灵订阅”的问题。4.2 事件顺序被破坏两个并发事件的竞争条件异步派发有一个隐藏问题事件的入队顺序和业务因果顺序不一定一致。我遇到过这样一个场景库存资源先扣减成功随后订单资源更新为已支付业务希望代理按“先库存后订单”的顺序处理实际收到的却是“订单先、库存后”。原因出在资源写入路径上。两个资源分别由不同模块维护模块 A 先写入库存并 emit模块 B 后写入订单并 emit但模块 B 的同步代码因为恰好赶上另一个异步任务比模块 A 更早进入 pending 队列。微任务队列只保证“谁先入队谁先出队”不保证“业务上谁先发生谁先入队”。我给这个场景补了事件序号机制。每个事件携带全局递增的sequence字段需要有序消费的代理维护一个lastHandledSequence只处理序号更大的事件。但这解决的是同一资源的事件顺序跨资源因果顺序仍然需要业务层显式表达。最终的做法是引入“屏障事件”。订单支付完成后业务层显式发出一个order.paid屏障事件库存扣减和积分累计的下游代理都要等待屏障事件到达后再执行。屏障事件不承载复杂数据只表示“前面的因果链条已经结束现在可以安全处理了”。把这个机制写清楚之后我们项目里再没发生过跨资源顺序错乱的事。4.3 重复回调幂等设计缺失导致重复扣减异步派发一个绕不过去的问题是失败重试。某个代理在处理库存扣减事件时因为数据库连接超时失败了REA 的重试策略会重新派发同一个事件。如果业务处理逻辑没有幂等保护第二次执行就会再次扣减库存。这个坑的教训不是“不要重试”而是“重试之前必须确认处理是可重入的”。REA 在框架层做了一件事为每个事件生成event.id sourceResourceId : sequence并提供已处理事件 ID 的校验工具。业务代理在入口处先检查这个 ID 是否处理过处理过就跳过。function handleDeductStock(event: DomainEvent, context: ReaContext) { const eventId ${event.sourceResourceId}:${event.sequence}; if (context.isHandled(eventId)) { return; } // 执行实际扣减逻辑 deductStock(event.meta.amount); context.markHandled(eventId); }这里有个顺序细节先扣减后markHandled还是先markHandled后扣减我建议先执行业务操作成功之后再标记。如果先标记再执行业务操作失败时标记已经写入重试就永远不会来了。当然如果业务系统有可靠的事务保证可以把扣减和标记放在同一事务里那是更优解。幂等标记不是万能药。跨服务、跨数据库的分布式环境下单纯靠内存集合做isHandled是不够的需要把标记写入持久化存储。REA 提供的是抽象接口默认实现是内存 Set线上则换成了数据库。关键在于业务层要理解这个工具的存在并且主动使用。5. REA 的性能表现与可靠性测试5.1 压测数据100 万事件场景下的内存与耗时写小轮子最怕性能翻车。我在内部测试环境做了一轮压测环境是普通开发机模拟注册了 100 个代理每个代理关注两个事件类型。然后连续制造 100 万次资源版本变化每次变化都触发对应事件。结果总耗时在 1.6 到 2.0 秒之间内存新增约 28 到 36 MB。这个数据算不上严谨 benchmark核心派发路径只验证了两件事一是如果不用深比较单纯靠 version 和事件类型索引单次派发开销可以压到微秒级二是事件对象和 pending 队列是内存增长的主要来源业务代理的临时状态往往占比更大。我特别看了一下内存分配的大头事件对象每次派发都会新建meta载荷越大内存上涨越快。后来针对高频场景做了小优化pending队列里不再直接存完整事件对象而是存事件的构造参数元组flush 的时候再组装成DomainEvent。这让内存峰值下降了大约 20%代价是代码可读性变差了一点。如果业务量没到百万级不建议做这个优化。5.2 靠重试和超时兜底Agent 的保障策略REA 提供一个withPolicy包装器用来给普通代理加上可靠性策略。我实际常用的配置如下const protectedAgent withPolicy(orderAgent, { retry: 3, timeoutMs: 5000, exponentialBackoff: true, });withPolicy内部做的事情不复杂调用onEvent时如果返回的 Promise 在超时时间内没有 settle就视为失败失败后按照指数退避策略重试重试次数耗尽仍失败就把事件写入死信队列并在日志里输出完整上下文。死信队列是排查问题的金矿。记录的内容至少要包含事件 ID、来源资源、代理 ID、失败次数、错误堆栈、事件发生时间。没有这些信息重试失败后就是一团迷雾只能靠翻日志碰运气。重试和幂等必须成对出现。没有任何保障策略时重试只会放大错误。我在接入withPolicy之前先让所有敏感代理都做了幂等改造然后才允许开启重试。顺序反过来线上一定会出重复处理的故障。6. 往分布式方向的扩展和现在的收尾6.1 从单机进程内到跨服务代理语义怎么变项目中期有别的团队想把 REA 用在跨服务场景我当时差点直接把注册表改成远程调用。冷静下来分析后发现“本地事件代理”和“分布式消息”的语义差异非常大强行复用只会让两边都别扭。差异主要体现在四个方面。第一注册表从进程内 Map 变成了消息队列的订阅关系不能再靠对象引用找到代理。第二事件从内存对象变成需要序列化的消息结构必须带上 schema 版本否则老消费者读到新字段会直接崩。第三派发时机从微任务变成了网络投递原来“同步收集、异步派发”的折叠效应完全失效。第四顺序保证从内存序号变成了分区有序跨分区仍然没有全局顺序。于是我们做了一个折中选择REA 继续保留进程内实现但在边界上增加一个适配器负责把本地事件转换成远程消息发送给已有消息通道同时监听远程消息并转换为本地事件。REA 不是消息中间件的替代品而是消息中间件和业务逻辑之间的翻译层。这样既保护了现有代码又让跨服务的事件流转能复用消息通道已有的可靠性能力。6.2 一些个人的体会从写下第一行原型到稳定运行REA 这个代号在项目里已经用了很长一段时间。我自己最大的体会是事件代理这种东西千万不要等到架构完全定了再动手。最好的方式是从业务里挑两三个真实卡点先写一个最简版本让它跑起来解决实际问题然后再根据问题反推设计。第一版大概率不是通用组件但它一定是最贴合真实场景的那一版后面的通用能力都是在这种“贴合”之上长出来的。如果你准备在自己的项目里做类似的东西我只有一个建议先把资源、事件、代理的边界画清楚再去写任何代码。边界清楚了后面所有的踩坑其实都有规律可循。
