本地事件分发这个能力乍一听像是框架里才有的东西其实只要项目体量稍微大一点迟早会需要它。我用过的几个中小型 Java 服务起初都是类与类之间直接调方法改到最后发现下单流程里塞了二十多行顺便通知一下积分顺便写条消息顺便刷新下缓存一个方法改三次每次都要重新回归全部下游逻辑。后来把事件总线引进来主流程只负责发一条事件谁关心谁自己订阅改动范围立刻收窄到一个订阅类里。这篇就围绕用 Java 自己实现一套本地事件分发展开讲清楚为什么要自己写、核心模型怎么设计、同步与异步怎么取舍、异常与顺序怎么兜底以及上线之后怎么排查那些诡异的问题。本地事件分发说白了就是在同一个 JVM 进程内把某件事发生了这个消息交给零个或多个感兴趣的接收者发布方不关心也不等待具体是谁处理的。它和消息队列最大的区别就是不出进程没有网络、没有序列化、没有中间件代价是可靠性由你自己兜底。适合读这篇文章的人正在被循环依赖和臃肿 Service 折磨的后端开发者、想给单体应用做模块解耦的架构同学以及准备自己动手写一个轻量组件而不是直接引入重型框架的人。下面所有代码和方案都是我在实际项目里跑过的形态不追求大而全追求能抄走就用。1. 先想清楚为什么不用现成的消息中间件1.1 进程内通信和跨进程通信是两类问题很多人一提事件就想到 MQ其实这两者的适用场景差别很大。我把判断标准压成一句话事件的消费者和生产者是否必须独立部署、独立扩容、独立故障隔离。如果答案是是那必须用跨进程的消息中间件如果答案是否用本地事件分发就够了而且简单得多。举个具体例子。用户注册成功后要做四件事发欢迎站内信、初始化积分账户、写一条操作审计日志、给推荐人加成长值。这四个动作全部在同一个应用里用的是同一个数据库连接池任何一个失败都不应该影响注册主流程的成功返回。这种场景就是本地事件分发的典型用法。如果你非要上 MQ得额外维护 broker、处理消息序列化版本、考虑消息重复投递的幂等运维和开发成本翻好几倍而收益几乎为零——因为消费者本来就跟生产者同生共死。我自己总结了一张对照表每次评审方案时直接拿出来用维度本地事件分发跨进程消息中间件通信边界同一 JVM 内跨进程、跨机器可靠性内存态进程挂则丢可持久化支持重投延迟纳秒到微秒级毫秒级起事务一致性可与本地事务同生共死需要最终一致性设计运维成本几乎为零需要部署与监控适用规模单体、模块化单体微服务、跨团队注意本地事件分发不等于不可靠。进程内不丢指的是同步分发时异常能立刻被感知异步分发时队列在内存里进程崩溃确实会丢这一点在方案选型时必须跟业务方讲清楚不要含混过去。1.2 事件驱动真正解决的是依赖方向问题代码腐烂往往不是因为功能复杂而是因为依赖方向混乱。订单服务依赖积分服务积分服务依赖消息服务消息服务又反过来查订单——一圈下来谁也不敢动。事件驱动最核心的价值不是异步而是把双向依赖掰成单向。没引入事件之前调用方必须持有被调用方的引用于是调用方知道被调用方的存在被调用方的方法签名一改调用方就得跟着改。引入事件之后调用方只知道有一个事件类型订阅方只知道我关心这个事件类型双方通过事件这个中间契约通信互相不持有引用。新增一个消费者时主流程代码一行都不用改这就是开闭原则落地最直接的样子。我见过一个比较极端的反例有个项目的下单方法里串行调用了十一个下游服务每个都做了 try-catch 吞异常。结果某次上线后积分没加、优惠券没核销日志里干干净净——因为异常全被吞了谁也不知道。改造成事件分发之后每个订阅方自己处理自己的异常并打日志问题定位时间从半天缩短到十分钟。这就是解耦带来的可观测性红利。1.3 自己写还是引框架边界在哪市面上有成熟的事件总线实现功能齐全但也带来一个问题你引入了一个可能只用 20% 功能的依赖还要跟着它的版本升级走。我自己动手写的原因有三个一是核心逻辑不到三百行维护成本可控二是可以完全贴合自己项目的异常策略和线程模型三是不用担心框架的内部实现变化导致行为漂移。不过有几个信号出现时我建议直接用成熟框架需要事件持久化、需要跨节点广播、需要事件重放、需要复杂的路由规则按标签、按优先级树形分发。这些能力自己写会越写越大最终变成半个中间件得不偿失。判断标准很朴素如果你的需求能用一张接口加一个 Map 表达就自己写如果开始需要配置文件描述路由就换框架。2. 把事件总线拆成四个最小零件2.1 事件对象不可变是第一原则事件对象的唯一职责是承载发生了什么它必须是不可变的。原因很实在同一个事件可能被多个订阅者同时拿到如果某个订阅者顺手改了字段后面的订阅者看到的数据就是脏的而这种 bug 极难复现。我用 Java 17 的 record 来定义天然不可变省掉一堆样板代码public record OrderCreatedEvent( Long orderId, Long userId, BigDecimal amount, Instant occurredAt ) { public OrderCreatedEvent { Objects.requireNonNull(orderId, orderId); Objects.requireNonNull(userId, userId); } }如果你的项目还在 Java 8用 final 字段加全参构造、只提供 getter 即可。关键是不要给事件加 setter也不要往里塞可变集合。如果确实需要传列表用List.copyOf()复制一份再放进去。事件里该放什么我的经验是放标识加必要快照不要放整个实体对象。放实体对象有两个坏处第一实体通常是可变的破坏了事件的不可变性第二实体可能带懒加载关联异步处理时 session 已关闭一访问就抛异常。所以正确做法是放 orderId 加上金额、用户 ID 这类计算需要的字段订阅方需要更多数据时自己去查。2.2 订阅者接口泛型加一个方法就够订阅者接口越简单越好一个方法一个泛型参数避免设计成万能处理器public interface EventListenerT { void onEvent(T event); default ClassT supportType() { ... } }这里有个 Java 泛型的老坑必须提运行时泛型被擦除你没法直接通过T拿到 Class 对象。解决方案有两种一种是让订阅者在构造时显式传 Class 进来另一种是用 Spring 的GenericTypeResolver之类工具反射解析父接口的泛型参数。我自己写的时候选了更土但更可靠的办法——在注册时由调用方显式指定事件类型虽然多写一个参数但永远不会出意外public interface EventListenerT { void onEvent(T event) throws Exception; } public final class EventBus { public T void subscribe(ClassT eventType, EventListener? super T listener) { ... } }EventListener? super T这个写法是有意为之的它允许一个EventListenerObject订阅所有具体类型的事件也就是所谓的通配订阅者用来做统一日志或监控特别方便。如果写成EventListenerT这个能力就没了。2.3 注册表类型到监听器列表的映射注册表本质就是一个ConcurrentHashMapClass?, ListEventListener?但细节决定成败。我踩过的坑集中在两点并发注册和遍历时修改。先说并发注册。应用启动阶段注册监听器通常是单线程的但有些项目会做动态注册比如插件式加载这时候就可能并发写。用ConcurrentHashMap的computeIfAbsent可以保证原子性但要注意返回的 List 本身也得是线程安全的。我一开始用了ArrayList结果在某次压测中偶发ConcurrentModificationException排查了半天才想起来是动态注册和事件分发撞在了一起。后来换成CopyOnWriteArrayList问题消失。为什么选CopyOnWriteArrayList而不是Collections.synchronizedList因为注册表的读远多于写——事件分发的频率可能是注册频率的几万倍而CopyOnWriteArrayList的读是无锁的只在写时复制整个数组。写操作虽然贵但发生得少完全划算。不过要提醒一句如果监听器数量上千且频繁动态增删CopyOnWriteArrayList的复制成本会变得明显那种场景就得换别的结构了但绝大多数业务系统到不了这个量级。遍历时修改的坑更隐蔽。分发事件的过程中某个监听器触发了新的注册动作如果直接遍历原始 List 就会抛异常。CopyOnWriteArrayList的迭代器是快照式的天然规避了这个问题——它遍历的是创建迭代器那一刻的数组副本新增的监听器下次分发才生效。这个语义我认为是合理的一次事件分发应该有一个确定的接收者集合。2.4 分发器同步与异步的分水岭分发器是整条链路的中枢它的核心逻辑只有三步按事件类型查监听器、依次调用、处理异常。但依次调用这四个字里藏着同步异步的选择这是整个设计里最需要想清楚的地方。同步分发的语义是发布方法返回时所有监听器都执行完毕。好处是简单、可预测、异常能立刻冒泡能和调用方的事务绑在一起。坏处是任何一个慢监听器都会拖慢主流程一个监听器抛异常会影响后续监听器除非你显式捕获。异步分发的语义是发布方法立刻返回监听器在线程池里慢慢跑。好处是不拖主流程坏处是异常拿不到、顺序不保证、进程崩溃会丢事件、还要处理线程池打满的降级策略。我的实际做法是两种都支持由订阅方自己声明而不是全局一刀切。这个设计很关键因为不同监听器的诉求完全不一样发站内信可以异步写审计日志最好同步要保证和业务操作在同一个事务里刷新本地缓存必须异步且要能容忍丢失。3. 同步分发的实现细节与事务边界3.1 主干代码六十行讲完同步分发的主干逻辑我写得尽量直白不搞花活public final class EventBus { private final MapClass?, ListEventListener? registry new ConcurrentHashMap(); public T void subscribe(ClassT type, EventListener? super T listener) { registry.computeIfAbsent(type, k - new CopyOnWriteArrayList()).add(listener); } SuppressWarnings(unchecked) public T void publish(T event) { Class? type event.getClass(); ListEventListener? listeners registry.get(type); if (listeners null || listeners.isEmpty()) { return; } for (EventListener? raw : listeners) { ((EventListenerT) raw).onEvent(event); } } }注意这里用的是event.getClass()精确匹配而不是instanceof遍历所有类型。为什么精确匹配的性能是 O(1) 哈希查找而遍历判断是 O(n)。代价是不支持父类型订阅——如果你 publish 一个OrderCreatedEvent只订阅了DomainEvent的监听器收不到。这个取舍我认为可以接受因为领域事件本来就该按具体类型订阅父类型订阅容易让人写出什么都收的监听器反而模糊了职责。如果确实需要支持继承关系我建议在注册阶段做索引展开注册DomainEvent时把当前已知的所有子类型都映射到它同时提供手动注册子类型的入口。但这会引入新类型注册前已有监听器收不到的一致性问题复杂度陡增我一般不推荐。3.2 异常策略必须是显式的同步分发里最容易被忽略的就是异常处理。默认行为是一个监听器抛异常后面的全不执行异常传给发布方这个行为合理吗取决于业务。我的做法是在 EventBus 里统一捕获并转成运行时异常同时保证所有监听器都被尝试执行一遍然后把异常聚合起来抛出去。public T void publish(T event) { ListEventListener? listeners registry.get(event.getClass()); if (listeners null || listeners.isEmpty()) return; ListThrowable errors new ArrayList(); for (EventListener? raw : listeners) { try { ((EventListenerT) raw).onEvent(event); } catch (Throwable t) { errors.add(t); log.error(listener failed, event{}, listener{}, event, raw.getClass().getName(), t); } } if (!errors.isEmpty()) { throw new EventDispatchException(分发失败 errors.size() 个监听器, errors); } }为什么选择全部执行完再抛而不是快速失败因为同步订阅的场景里监听器之间通常互相独立写审计、刷缓存一个失败不应该导致另一个也不执行。如果确实有强依赖顺序的监听器那不是事件该解决的问题应该拆成一个编排服务。自定义异常EventDispatchException里带上ListThrowable方便上层统一处理或者只记录不中断。这里有个细节捕获的是Throwable而不是Exception因为要防止某个监听器抛Error导致整个分发链条静默中断。当然捕获Throwable后不要吞掉OutOfMemoryError这类致命错误我的做法是对Error单独判断记录后重新抛出。3.3 事务边界什么时候发事件决定了数据一致性这是本地事件分发最容易出事的地方我单独拎出来讲。假设用户注册流程是插用户表事务内、发注册事件触发积分初始化。如果事件在事务提交前发布监听器里也操作数据库用的是同一个事务那没问题要么全成功要么全回滚。但如果监听器是异步的它在事务提交前就开始跑了可能读到还没提交的数据取决于隔离级别也可能主事务回滚了但异步任务已经把积分加了——数据不一致。所以我的原则非常明确同步监听器可以在事务内发布异步监听器必须在事务提交后发布。实现上依赖 Spring 的TransactionSynchronizationManager在afterCommit回调里触发异步分发public void publishAfterCommit(Object event) { if (TransactionSynchronizationManager.isSynchronizationActive()) { TransactionSynchronizationManager.registerSynchronization( new TransactionSynchronization() { Override public void afterCommit() { eventBus.publishAsync(event); } }); } else { eventBus.publishAsync(event); } }这段代码看着简单但它解决的是一个非常实际的问题。我见过有项目在事务里发异步事件结果主事务因为唯一键冲突回滚了但下游的站内信已经发出去了用户收到注册成功的站内信却发现自己登录不上客服电话直接打爆。加上afterCommit之后这类问题彻底消失。提示afterCommit里抛异常不会影响主事务因为已经提交了所以异步任务提交线程池失败时要在那里做好日志和补偿别指望异常能回滚什么。3.4 监听器顺序需要时排序不需要别排序有些场景确实需要顺序比如先校验再落库。我提供两种机制一是注册时传一个 order 值分发前按 order 排序二是用Order注解配合反射读取。前者更直接后者更符合 Spring 风格。排序有个性能代价每次分发都排序太浪费。我的做法是在注册时维护一个有序结构用一个volatile的不可变列表缓存排序结果注册动作发生时重新构建这个列表。这样分发路径上一次排序都不用做纯遍历。private volatile ListEventListener? orderedCache; private void rebuildCache(Class? type) { ListEventListener? list new ArrayList(registry.get(type)); list.sort(Comparator.comparingInt(this::orderOf)); this.orderedCache List.copyOf(list); }不过要提醒顺序只在同步分发里有意义。异步分发时多个任务在线程池里并行跑顺序完全不可控这时候纠结 order 值是自欺欺人。如果业务真的需要异步且有序那只有两个选择单线程执行器牺牲吞吐或者干脆改成同步。这一点在设计阶段就要跟业务方对齐别等到上线才发现顺序错了。4. 异步分发线程池是唯一的难点4.1 线程池参数怎么定别拍脑袋异步分发本质就是把任务丢给线程池难点全在参数上。核心线程数、最大线程数、队列容量、拒绝策略这四个参数配错轻则延迟飙升重则整个应用被拖垮。我的取值逻辑是这样的这类任务是 IO 密集型写库、发通知、调外部接口按经验公式核心线程数 CPU 核数 × 2起步。但这个公式只是起点真正定值要看压测。我一般的做法是先给一个保守值比如 8然后观察队列的堆积情况和任务执行耗时用监控数据反推。队列容量我踩过的坑最深。曾经图省事用了LinkedBlockingQueue无界队列结果下游某个接口变慢任务堆积了几十万个内存直接飙到 OOM。后来改成有界队列容量按峰值 QPS × 可接受延迟秒数估算。比如峰值每秒 500 个事件能容忍延迟 10 秒那队列容量就是 5000 左右留点余量给 8000。拒绝策略是最后一道防线。AbortPolicy会抛异常同步感知到但异步场景里调用方往往是业务主流程抛异常会打断它CallerRunsPolicy让提交任务的线程自己执行起到天然限流的作用但会拖慢调用方DiscardPolicy静默丢弃最危险。我的选择是自定义策略记录日志 打点计数 丢弃并且把丢弃计数暴露到监控面板上超过阈值就告警。因为对大多数非关键通知类事件来说丢一条比拖垮主流程好。4.2 上下文传递ThreadLocal 的坑等着你异步分发最容易被忽略的问题就是上下文丢失。主线程里的ThreadLocal用户身份、链路追踪 ID、租户 ID到了线程池的工作线程里全是空的。刚开始我没注意结果审计日志里的操作人全是 null排查了好久。解决方案是在提交任务时抓取上下文快照执行时恢复执行完清理。Spring 有TaskDecorator可以挂在线程池上做这件事public class ContextCopyingDecorator implements TaskDecorator { Override public Runnable decorate(Runnable task) { MapString, String ctx MdcContext.snapshot(); String traceId TraceContext.current(); return () - { try { MdcContext.restore(ctx); TraceContext.set(traceId); task.run(); } finally { MdcContext.clear(); TraceContext.clear(); } }; } }finally里的清理绝对不能省。线程池的线程是复用的如果上一个任务留下的上下文没清掉下一个任务就会读到一个别人的用户身份这种 bug 极其隐蔽可能几个月都没人发现最后表现为某用户的审计日志记到了别人头上。另外提醒一点不要把整个HttpServletRequest或者大对象塞进上下文快照里只抓那几个字符串字段。我见过有人把整个请求上下文都存进去结果线程池里的任务持有请求对象引用导致请求对象无法回收内存泄漏。4.3 背压当生产者比消费者快得多异步分发的隐患是生产速度超过消费速度时队列无限增长。有界队列 拒绝策略能兜住内存但兜住的方式是丢任务业务上未必能接受。这时候需要考虑背压也就是让生产方感知到压力并主动降速或降级。我的做法分三个层次。第一层是监控把队列长度、活跃线程数、拒绝次数暴露出去先做到看得见。第二层是阈值告警队列使用率超过 70% 就报警给运维留出反应时间。第三层才是业务降级比如在队列接近满的时候把非核心事件如运营统计直接丢弃只保留核心事件如账户变动。如果业务对丢失零容忍那本地事件分发就不合适了应该老老实实上持久化队列。这个判断必须诚实不要因为已经写了一版本地实现就硬撑着往里加持久化最后写出来的东西是半个残缺的中间件还不如直接用成熟的。5. 那些只有上线后才会暴露的问题5.1 递归分发导致栈溢出监听器里再发事件是很自然的写法比如订单创建事件的监听器发一个积分变动事件。正常情况没问题但如果形成环——A 事件触发 B 事件B 事件又触发 A 事件——就会无限递归最后StackOverflowError。我在测试环境就遇到过这种事日志里刷了几千行然后服务挂掉。修复方案是加一个分发深度限制用一个ThreadLocalInteger记录当前调用深度超过阈值比如 5 层就记录警告并拒绝分发private static final ThreadLocalInteger DEPTH ThreadLocal.withInitial(() - 0); private static final int MAX_DEPTH 5; private void checkDepth() { int d DEPTH.get(); if (d MAX_DEPTH) { throw new IllegalStateException(事件分发深度超限疑似循环依赖: d); } DEPTH.set(d 1); try { doPublish(event); } finally { DEPTH.set(d); if (d 0) DEPTH.remove(); } }这个机制的价值不只是防栈溢出更重要的是它能在测试阶段把循环依赖暴露成一个明确的异常而不是让它在线上随机爆发。我在项目里还加了一个开发环境的开关开启后把每次分发的链路事件类型序列打出来配合深度限制环在哪一环一目了然。5.2 慢监听器拖垮整个线程池异步线程池里的任务耗时不一致是很常见的。发站内信可能 10ms但调外部短信接口可能 3 秒还超时。几个慢任务就能把线程池占满导致后面的快任务排队。解决办法是按耗时隔离线程池把慢任务和快任务分开各自用独立的有界线程池。这样慢任务再慢也影响不到快任务。隔离的粒度按事件类别划分比如通知类一个池、统计类一个池强一致类走同步不走池。另外必须给监听器设超时。我用Future.get(timeout)的方式包一层太笨重实际做法是在线程池层面配合业务层面的超时控制——外部调用本身必须配超时HTTP 客户端连接超时、读取超时数据库查询超时这是根本。如果外部调用不设超时你在线程池层面怎么隔离都会被堵死。5.3 监听器的生命周期管理注册容易注销难大部分项目只关心注册不关心注销。但如果你的监听器是动态的——比如按插件加载、按租户创建、按配置开关——就必须有注销机制。没有注销的后果是监听器持有对象引用对象无法回收内存泄漏或者功能下线圈后监听器还在跑产生意料之外的副作用。我给EventBus加了一个unsubscribe方法注册时返回一个Subscription句柄调用handle.close()就注销public interface Subscription extends AutoCloseable { Override void close(); }实现上就是记住注册时的类型和监听器实例close()时从列表中移除。配合 try-with-resources 用起来很顺手尤其适合测试代码——每个测试方法里注册的监听器在结束时自动注销不会污染其他测试。6. 怎么验证这套东西真的能用6.1 单元测试要覆盖的五个场景自己写的组件测试得自己兜住。我列了一张必测清单每次改动都跑一遍场景验证点无监听器时发布不抛异常正常返回单监听器正常处理收到事件字段值正确多监听器部分抛异常全部被调用异常被聚合抛出通配监听器收到所有类型的事件并发发布 并发注册无异常无漏发并发那个场景要用多线程跑重点看有没有ConcurrentModificationException和事件漏发。我一般用CountDownLatch起 8 个线程各发 1000 次注册线程同时增删监听器跑完检查计数是否对得上。这类测试偶尔过、偶尔失败恰恰说明并发问题真实存在必须跑到稳定为止。6.2 一个容易被忽略的验证点事件对象确实不可变不可变性这种东西靠 code review 靠不住得靠工具。要么在事件类上加Immutable之类的注解配合静态检查要么写一个测试用反射把事件对象的所有字段拿出来尝试构造一个修改后的对象去比较哈希值。更直接的办法是让事件类只有 record 或只有 final 字段从语言层面杜绝。还有一个实践在事件分发前后各算一次hashCode()如果变了就说明有监听器动了手脚直接抛异常。生产环境我不建议开这个检查有性能成本但在测试环境打开非常有价值能抓到那些顺手改一下的代码。6.3 压测关注什么指标本地事件分发的压测不看吞吐量绝对值重点看三件事分发链路的 P99 延迟、线程池队列的堆积速度、拒绝次数和异常次数。同步分发关注的是监听器总耗时占主流程的百分比——如果某个监听器占了主流程 40% 的时间那它就该改成异步。异步分发关注的是队列消费速度和堆积趋势如果堆积持续上升说明消费能力不足。我通常在压测报告里明确列出每个监听器的耗时这样谁慢谁快一目了然优化方向也清楚。7. 一点选型之外的体会自己写事件总线这件事最容易被低估的不是代码量而是语义定义的清晰度。同步还是异步、顺序有无保证、异常怎么传播、事务边界在哪、丢不丢事件——这几个问题在写代码之前回答得越清楚后面的坑越少。我第一版就是因为没想清楚事务边界上线后遇到了一次数据不一致回头补afterCommit花的时间比写整个组件还多。还有一点不要因为有了事件总线就滥用。我见过把每个方法调用都改成事件的写法结果调用链完全不可见想知道注册成功之后到底发生了什么得在代码里搜所有订阅者调试成本极高。事件适合用在一对多、且订阅方会持续增加的场景一对一的直接调用老老实实写直接调用别为了解耦而解耦。最后分享一个小技巧在EventBus里加一个开发环境的分发日志开关打开后把每次分发的事件类型、监听器列表、耗时都打到日志里。这个开关我几乎在每次排查问题时都会打开它的价值在于把一条隐式的调用链重新变成显式的尤其是排查为什么这个操作没触发那个副作用这类问题时一眼就能看出是没发事件、还是没人订阅、还是订阅了但抛异常被吞了。