★ FEATURED ARTICLE 简介面向希望深入理解Guava事件总线机制的Java开发者这份学习资料以Guava 18.0版本的EventBus源码为分析对象聚焦发布/订阅模型如何解耦生产者与消费者以及同步、异步事件传递中的线程控制思路适合具备一定Java基础、准备阅读框架源码或在实际项目中评估事件驱动方案的中级开发者参考。资源包共26个文件整体约775KB主体为15个Java源码文件并配有6张流程示意图、3份UML图以及说明文档和构建配置便于将类结构、事件注册、发布与注销流程对照起来理解而不是停留在零散API调用层面。目前已有1898人学习。内容围绕EventBus与AsyncEventBus的差异、Executor线程控制、并发执行注解等使用约束展开配合图示和UML可帮助读者梳理事件总线核心链路形成从源码结构到设计取舍的完整笔记适合作为框架源码入门与事件驱动设计的参考资料。1. 从一处ListXxxListener说起EventBus 到底省掉了什么老工程里常见这样一段代码用户注册成功后要发积分、发短信、写审计日志于是register()方法里塞了三个xxxListener.onRegistered()。以后每加一项业务都得回来改这段代码。Guava 的 EventBus 把这根线调了个方向谁关心谁自己挂一个带Subscribe的方法发事件的人只负责bus.post(event)剩下的交给注册表。这份资源就是把 Java 生态里com.google.common.eventbus这个不到三千行的包按类拆开配上能直接跑的 Demo 和源码注释。适合两类人用过 Spring 的EventListener但没往下看过实现的以及想给自己框架加一条进程内事件总线、又不想引入整套消息中间件的。下面按先看骨架、再跑起来、最后自己改的顺序来。2. 拆包先看骨架com.google.common.eventbus里的类分工读任何源码包我都先做一件事把目录树打印出来看文件名猜职责画好职责边界再进代码。EventBus 这个包的好处是类不多、边界清楚一个下午能过一遍坏处是几个关键设计藏得比较深比如为什么注册中心用CopyOnWriteArraySet、为什么分发器要挂ThreadLocal这些不看代码只看文档是猜不出来的。2.1 11 个类各管一件事先把清单摆出来。这份资源的结构基本就是按这个清单组织的每节对应一到两个类读的时候可以对着看。类一句话职责读源码时要盯的点EventBus门面register/unregister/postpost只有三行分支但分支里藏着 DeadEventAsyncEventBus构造时注入Executor的子类和父类的差异只有分发器和线程池SubscriberRegistry事件类型到订阅者集合的注册中心ConcurrentMap加CopyOnWriteArraySet的组合Subscriber一个「监听器实例 Method」的组合体反射调用与异常包装SynchronizedSubscriber给非线程安全的订阅方法加锁AllowConcurrentEvents的反面Dispatcher抽象调度器决定回调时机两个实现就是同步与异步的分水岭PerThreadQueuedDispatcher每线程一条队列递归post不爆栈的原因LegacyAsyncDispatcher全局并发队列回调顺序在这里没保证DeadEvent无人认领事件的包装壳排查「发了没反应」的第一现场Subscribe/AllowConcurrentEvents标记注解保留策略、方法签名约束SubscriberExceptionHandler/SubscriberExceptionContext异常出口与上下文默认实现把异常写进日志这张表不是抄注释来的是我按「谁调用谁」重新排的。门面层只做转发注册层管映射执行层管调用异常层兜底四层之间几乎没有交叉引用这是它敢放进公共库的底气。2.2post的调用链四层一层不多真正把事件从发出到落地串起来的只有下面这四层。这是全篇最该记住的一段代码// EventBus.java节选保留主干 public void post(Object event) { IteratorSubscriber eventSubscribers subscribers.getSubscribers(event); if (eventSubscribers.hasNext()) { dispatcher.dispatch(event, eventSubscribers); } else if (!(event instanceof DeadEvent)) { // 没人认领包一层 DeadEvent 再投一次避免无限递归 post(new DeadEvent(this, event)); } }第一层getSubscribers拿的是事件类型对应的订阅者迭代器注意是Iterator不是List因为同一个事件可能有多个父类型各自挂着订阅者最终要给一个拼接后的视图。第二层dispatcher.dispatch决定「什么时候真的调」同步实现里它是直接就地遍历异步实现里它把活丢给线程池。第三层Subscriber.dispatchEvent才是反射调用入口第四层落到method.invoke。参数说明只有一句重要的event不允许为nulldispatch里会checkNotNull拦下来所以别在业务里用post(null)表示「什么都没发生」。这里有个容易忽略的设计DeadEvent的判断是instanceof而不是精确匹配。也就是说你如果自己 new 一个DeadEvent直接post进去它不会再去包装第二层否则就无限递归了。2.3 两个关键决策CopyOnWrite与ThreadLocal整个包的性能取向其实由两个数据结构定下来。第一个是注册表里的ConcurrentMapClass?, CopyOnWriteArraySetSubscriber。选它是基于一个假设注册发生在启动阶段一次几十上百个事件分发发生在运行期一天几十万次。读多写少的场景下CopyOnWriteArraySet让遍历完全无锁——遍历时拿的是数组快照其他线程怎么注册删除都不影响这次遍历。代价是每次register都复制一份数组所以别拿它做高频动态注册。第二个是PerThreadQueuedDispatcher里的两个ThreadLocal一个QueueEvent存待处理的事件一个Boolean表示「当前线程是否已经在分发循环里」。第二个标记才是精髓它把一个可能递归的调用拍平成了循环。提示理解这两个决策之后很多「为什么它不支持 X」的疑问会自己消失。因为它压根没打算做高频动态注册也没打算做跨线程顺序保证。3. 注册与分发源码注解怎么被扫出来、回调怎么被排队骨架看完进正题。这一章挑四个函数读register、findAllSubscribers、flattenHierarchy、dispatch。这四个函数加起来不到两百行但把 EventBus 的所有脾气都写进去了。3.1register三步把注解变成订阅者注册的入口在门面类里真正的活交给SubscriberRegistry。拆开看是三件事找到带注解的方法、包装成Subscriber、塞进按事件类型分组的集合。// SubscriberRegistry.java节选 public void register(Object listener) { // 第一步扫描这个实例含父类、接口默认实现里所有带 Subscribe 的方法 MultimapClass?, Subscriber listenerMethods findAllSubscribers(listener); for (Map.EntryClass?, CollectionSubscriber entry : listenerMethods.asMap().entrySet()) { Class? eventType entry.getKey(); // 事件类型 方法唯一参数的声明类型 CollectionSubscriber eventMethodsInListener entry.getValue(); // 第二步拿或原子地创建该事件类型对应的集合 CopyOnWriteArraySetSubscriber eventSubscribers subscribers.get(eventType); if (eventSubscribers null) { CopyOnWriteArraySetSubscriber newSet new CopyOnWriteArraySet(); eventSubscribers MoreObjects.firstNonNull(subscribers.putIfAbsent(eventType, newSet), newSet); } // 第三步整批加入注意 addAll 才触发一次数组复制 eventSubscribers.addAll(eventMethodsInListener); } }putIfAbsent加firstNonNull是标准的双检写法如果putIfAbsent返回旧的集合说明别的线程抢先建好了用旧的返回null说明自己赢了用新建的。写作时容易漏掉addAll这个细节——如果你把它换成循环add注册十个方法就复制十次数组启动阶段的耗时立刻翻倍。再看findAllSubscribers它是把注解解析成数据的那一层private MultimapClass?, Subscriber findAllSubscribers(Object listener) { MultimapClass?, Subscriber methodsInListener HashMultimap.create(); for (Method method : getAnnotatedMethods(listener.getClass())) { Class?[] parameterTypes method.getParameterTypes(); Class? eventType parameterTypes[0]; // 前面已经强校验过「有且仅有一个参数」 methodsInListener.put(eventType, Subscriber.create(bus, listener, method)); } return methodsInListener; }参数说明只有一个eventType取的是方法参数的声明类型不是运行时类型。所以订阅Object的方法会收到所有事件订阅CharSequence的方法会收到所有String事件的投递这个行为由 3.2 的flattenHierarchy决定不是在这里判断的。扫描方法本身还有个校验值得单独看一眼因为它解释了后面好几个坑for (Class? supertype : TypeToken.of(clazz).getTypes().rawTypes()) { for (Method method : supertype.getDeclaredMethods()) { if (method.isAnnotationPresent(Subscribe.class) !method.isSynthetic()) { Class?[] parameterTypes method.getParameterTypes(); checkArgument(parameterTypes.length 1, %s.%s 有 %s 个参数但 Subscribe 方法必须且只能有一个参数, clazz.getName(), method.getName(), parameterTypes.length); ... } } }注意checkArgument抛的是IllegalArgumentException发生在register阶段不是post阶段。也就是说参数写错了应用启动就会炸这是好事比运行期静默失败强得多。但private方法、static方法这里并没有显式拦截——getDeclaredMethods能拿到它们注解也在于是注册会成功调用时才在反射那里抛异常。这是第一个大坑第 5 章细说。3.2flattenHierarchy父类和接口是怎么被匹配上的订阅者和事件之间不是精确匹配而是「事件类型及其所有父类型」都能命中。实现只有一行核心逻辑TypeToken.of(clazz).getTypes().rawTypes()。这行会展开泛型信息后取所有原始类型包括自己、所有祖先类、所有实现过的接口。private static final LoadingCacheClass?, ImmutableSetClass? flattenHierarchyCache CacheBuilder.newBuilder() .weakKeys() // 弱引用键类被卸载时缓存条目自动失效不撑爆类加载器 .build(new CacheLoaderClass?, ImmutableSetClass?() { Override public ImmutableSetClass? load(Class? concreteClass) { return ImmutableSet.copyOf(TypeToken.of(concreteClass).getTypes().rawTypes()); } }); private static ImmutableSetClass? flattenHierarchy(Class? concreteClass) { try { return flattenHierarchyCache.getUnchecked(concreteClass); } catch (UncheckedExecutionException e) { throw Throwables.propagate(e.getCause()); // 老版本写法新版本已改读到这里别卡住 } }三个参数值得说清楚。weakKeys()是为了避免缓存强引用Class对象导致类加载器无法回收这在容器热部署场景里是血泪经验。缓存是因为层级展开走的是泛型反射一条链路下来开销不小而事件类型通常就那么几种缓存命中率极高。最后那个Throwables.propagate在新版本里已经被替换掉了读源码时看到它别以为是 bug只是历史包袱。展开成一个具体例子OrderCreatedEvent extends DomainEvent implements Traceable那么post(new OrderCreatedEvent())会依次查OrderCreatedEvent、DomainEvent、Object、Traceable这几个键的订阅者集合全部拼进同一个迭代器。顺序是「自己 → 父类 → 接口」这也是为什么给Object挂订阅者等于订阅了全世界。3.3PerThreadQueuedDispatcher递归post为什么不会爆栈这是整个包里最巧的一段。先想一个问题订阅方法里又post了一个新事件会发生什么最朴素的实现是直接递归调用分发层级一多就栈溢出PerThreadQueuedDispatcher的做法是排队。private static final class PerThreadQueuedDispatcher extends Dispatcher { private final ThreadLocalQueueEvent queue new ThreadLocalQueueEvent() { Override protected QueueEvent initialValue() { return Queues.newArrayDeque(); } }; private final ThreadLocalBoolean dispatching new ThreadLocalBoolean() { Override protected Boolean initialValue() { return false; } }; Override void dispatch(Object event, IteratorSubscriber subscribers) { checkNotNull(event); checkNotNull(subscribers); QueueEvent queueForThread queue.get(); queueForThread.offer(new Event(event, subscribers)); // 先入队 if (!dispatching.get()) { // 只有「最外层」那次才真正执行循环 dispatching.set(true); try { Event nextEvent; while ((nextEvent queueForThread.poll()) ! null) { while (nextEvent.subscribers.hasNext()) { nextEvent.subscribers.next().dispatchEvent(nextEvent.event); } } } finally { dispatching.remove(); // 一定要清否则线程复用时会带着脏状态 queue.remove(); } } } }dispatching这个标记的含义是「当前线程是否已经在分发循环里」。第一次进来是false设成true后进循环循环里某个订阅方法又post再次进到dispatch此时dispatching.get()是true于是只入队、立刻返回。外层循环poll到新事件时再处理它。整条链路深度永远是 1栈不会涨。这个设计带来两个可观测的行为写业务代码时得心里有数。其一事件处理顺序是「广度优先」而不是「深度优先」A 事件的订阅者发起的 B 事件要等 A 的所有订阅者跑完才会被处理。其二finally里必须remove否则线程池复用线程时会带着上一次的队列和标记出现「上一个请求的事件被下一个请求处理」这种玄学问题。Guava 这里写对了你自己实现的时候特别容易漏。AsyncEventBus用的是另一个实现LegacyAsyncDispatcher思路是把所有「事件 单个订阅者」的配对塞进一个ConcurrentLinkedQueue然后挨个dispatch真正的执行丢给Executor。名字里的Legacy就是提醒这是早期设计跨线程顺序不做任何保证。3.4Subscriber反射调用、同步包装与异常出口最底下这层负责真正把方法调起来同时决定了并发语义。static Subscriber create(EventBus bus, Object listener, Method method) { // 没标 AllowConcurrentEvents 就包一层同步壳保证同一个订阅者实例不会被并发进入 return isDeclaredThreadSafe(method) ? new Subscriber(bus, listener, method) : new SynchronizedSubscriber(bus, listener, method); } final void dispatchEvent(final Object event) { executor.execute(() - { try { invokeSubscriberMethod(event); } catch (InvocationTargetException e) { // 反射会把业务异常包一层这里剥掉外壳交给异常处理器 bus.handleSubscriberException(e.getCause(), context(event)); } }); }三个点。第一SynchronizedSubscriber是synchronized (this)锁在订阅者实例上粒度是「一个监听器对象」不是全局锁所以不同监听器之间不互斥。第二InvocationTargetException必须剥壳否则业务方拿到的堆栈里永远多一层反射帧排查时会被误导。第三异常不会往外抛——handleSubscriberException默认走LoggingHandler把异常写进日志就完了。这就是「订阅方法抛异常发事件那边完全不知道」的根源第 5 章会给出接住它的办法。注意AllowConcurrentEvents不是「允许并发」的开关而是「我确认自己是线程安全的别给我加锁」的声明。加错了不会报错只会在压测时表现为数据错乱。4. 跑起来一个可复现的最小工程源码读到这里验证一遍最快的方式是自己写三个 Demo同步、异步、异常可观测。这一章给出可直接抄的代码和每个参数的取舍理由。4.1 工程结构与依赖我一般这么摆一个模块放 Demo一个模块放源码笔记互不干扰eventbus-lab/ ├── pom.xml └── src/main/java/lab/eventbus/ ├── event/OrderCreatedEvent.java ├── demo/SyncDemo.java ├── demo/AsyncDemo.java ├── demo/DeadEventListener.java └── demo/BusExceptionHandler.java依赖只有一个版本别单独指定用工程里已有的${guava.version}dependency groupIdcom.google.guava/groupId artifactIdguava/artifactId version${guava.version}/version /dependency版本这块有个实际经验不要为了跑 Demo 单独升 Guava 大版本。它的小版本之间偶尔有方法签名调整比如前面提到的Throwables.propagate你工程里其他依赖可能锁着别的版本一升就冲突。用dependencyManagement统一管或者干脆跟随工程现状。4.2 同步版第一次看见Subscribe生效事件类保持不可变字段全final这是使用 EventBus 的基本纪律——异步场景下事件对象会被跨线程读取。// OrderCreatedEvent.java public final class OrderCreatedEvent { private final String orderId; private final long userId; private final long amountInCents; public OrderCreatedEvent(String orderId, long userId, long amountInCents) { this.orderId orderId; this.userId userId; this.amountInCents amountInCents; } // getter 略 }监听器和投递代码如下public class SyncDemo { public static void main(String[] args) { EventBus bus new EventBus(); // 默认构造同步分发 默认异常处理器 bus.register(new Object() { Subscribe public void addPoints(OrderCreatedEvent e) { System.out.println([积分] e.getOrderId() 线程 Thread.currentThread().getName()); } }); bus.register(new Object() { Subscribe public void sendSms(OrderCreatedEvent e) { System.out.println([短信] e.getOrderId()); } }); bus.post(new OrderCreatedEvent(ORD-1, 1001L, 9900L)); System.out.println(post 返回); } }跑出来的顺序是「积分 → 短信 → post 返回」两个监听器都在主线程里执行。这两个结论都要亲手验证一次因为它们是后面所有坑的基准同步 EventBus 里post是一个阻塞调用所有订阅方法串行执行完毕才返回。有人以为它是「发出去就完事」把耗时的短信调用放进去接口 P99 立刻起飞。注册顺序在这里等于回调顺序因为CopyOnWriteArraySet内部用的是数组addAll保持插入顺序。如果某天你把监听器改成从 Spring 容器里按Map迭代顺序注册回调顺序可能就变了——这是排查「为什么日志顺序不对」时的第一怀疑对象。4.3 异步版换线程池别换语义异步只需要两步构造AsyncEventBus并传线程池。public class AsyncDemo { public static void main(String[] args) throws Exception { // 有界队列 命名线程拒绝策略用 CallerRuns过载时降级到调用线程执行 ExecutorService pool new ThreadPoolExecutor( 4, 8, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue(1024), new ThreadFactoryBuilder().setNameFormat(eventbus-%d).build(), new ThreadPoolExecutor.CallerRunsPolicy()); EventBus bus new AsyncEventBus(pool); bus.register(new Object() { AllowConcurrentEvents // 无状态方法可以并发执行省掉同步壳 Subscribe public void slowHandler(OrderCreatedEvent e) throws InterruptedException { Thread.sleep(200); System.out.println([慢任务] e.getOrderId() 线程 Thread.currentThread().getName()); } }); for (int i 0; i 8; i) { bus.post(new OrderCreatedEvent(ORD- i, 1000L i, 100L)); } pool.shutdown(); pool.awaitTermination(5, TimeUnit.SECONDS); } }参数逐个说。核心线程 4、最大 8是按「订阅方法以 IO 为主」估的具体值看你订阅方法里干什么别照抄。队列必须有界LinkedBlockingQueue无界的话过载时内存先炸。线程名一定要起异步场景下日志里没有线程名基本没法排查。拒绝策略我习惯用CallerRunsPolicy队列满了让投递线程自己跑虽然会拖慢post但至少不会丢事件。注意这里用了AllowConcurrentEvents它让Subscriber不再包同步壳同一个监听器实例上的多个方法调用可以并发进。前提是你的方法真的无状态。判断标准很简单——方法里只读参数、不动成员变量就可以加。还有一个容易搞错的点异步版的回调顺序完全不做保证。LegacyAsyncDispatcher把「事件 订阅者」的配对塞进并发队列再派给线程池8 个事件打到 4 个线程谁先跑完谁先打印。所以别在异步版里依赖顺序要顺序就合并成一个处理器。4.4DeadEvent与异常处理器把静默失败变可见这两个是排查问题的抓手Demo 里一定要有。// 没有人订阅的事件会被包装成 DeadEvent 投回来 public class DeadEventListener { Subscribe public void onDeadEvent(DeadEvent deadEvent) { System.out.println([无人认领] event deadEvent.getEvent() source deadEvent.getSource()); } }getEvent()是原始事件getSource()是那个没人认领的 EventBus 实例。线上加这么一个监听器成本几乎为零但能立刻发现「事件发了没人接」这类问题——通常是包名写错、监听器忘了注册、或者事件类型对不上。异常处理器则用来兜住订阅方法里抛出来的任何东西public class BusExceptionHandler implements SubscriberExceptionHandler { private static final Logger log LoggerFactory.getLogger(BusExceptionHandler.class); Override public void handleException(Throwable exception, SubscriberExceptionContext context) { log.error(订阅方法执行失败 bus{} event{} subscriber{} method{} arg{}, context.getEventBus(), context.getEvent(), context.getSubscriber(), context.getSubscriberMethod().getName(), context.getEvent(), exception); } }上下文里最有价值的是getSubscriberMethod()它直接告诉你哪个类的哪个方法炸了比堆栈翻半天快得多。构造 EventBus 时把它传进去new EventBus(new BusExceptionHandler())。默认的LoggingHandler用的是 JDK 自带日志很多工程配不到它等于异常消失了。4.5 三条断言验证你对源码的理解Demo 跑通不代表理解对了。我习惯写三条测试钉住行为改成别的实现时也能用来对比Test public void 同步分发发生在调用线程内() { EventBus bus new EventBus(); AtomicReferenceString threadName new AtomicReference(); bus.register(new Object() { Subscribe public void on(String event) { threadName.set(Thread.currentThread().getName()); } }); bus.post(hello); assertEquals(Thread.currentThread().getName(), threadName.get()); } Test public void 父类型订阅者也能收到子类型事件() { EventBus bus new EventBus(); ListString hits new ArrayList(); bus.register(new Object() { Subscribe public void onObject(Object event) { hits.add(object); } Subscribe public void onString(String event) { hits.add(string); } }); bus.post(hello); assertEquals(Arrays.asList(object, string), hits); // 顺序取决于类型展开顺序 }第二条断言的顺序不一定和你本地跑的一致因为类型展开走的是TypeToken实际顺序取决于类和接口的排列。这里要验证的是「两个都收到」而不是具体顺序如果你需要确定顺序那说明用法本身有问题。第三条一般验证 DeadEventpost一个没人订阅的类型断言DeadEventListener收到了。这三条写下来你对 2.2 那条调用链的记忆会牢得多。5. 避坑与排查五个我踩过的 EventBus 坑源码看懂了线上照样翻车。下面五条都是「代码看起来完全没问题」的那种每条按现象、原因、解决写。5.1 注册成功了但事件永远收不到现象日志里register正常返回post那边也正常就是订阅方法不执行也没有 DeadEvent。原因方法写成了private或static。3.1 那段扫描代码只校验了「参数必须只有一个」没有校验方法修饰符所以注册阶段一路通过Subscriber里反射invoke时抛IllegalAccessException——但要注意dispatchEvent只 catch 了InvocationTargetExceptionIllegalAccessException是受检异常在invokeSubscriberMethod里被包成了运行时异常往上冒最后要么被吞要么出现在日志的某个角落。解决Subscribe方法一律public、非静态、非抽象、返回值void、参数恰好一个。把这条规则写进代码规范或者注册时自己加一层校验扫描一遍Modifier.isPublic(method.getModifiers())。5.2 订阅方法抛了异常事件像凭空消失现象业务方说「下单了但积分没加」日志里搜不到任何异常。原因Subscriber.dispatchEvent里catch (InvocationTargetException e)之后调handleSubscriberException默认处理器把异常写进 JDK 日志而很多工程的日志框架没接管java.util.logging于是异常写进了一个没人看的文件。加上post是同步调用但异常不往外抛上游完全无感知。解决构造 EventBus 时显式传自己的SubscriberExceptionHandler见 4.4至少log.error带上下文。更严格一点可以在处理器里打点上报按订阅方法名做监控。5.3 异步版里数据错乱同步版复现不了现象AsyncEventBus下同一个监听器的成员变量被写乱HashMap出现死循环或者计数不对切回同步 EventBus 一切正常。原因监听器有状态但没有标AllowConcurrentEvents——不对要反过来看标了AllowConcurrentEvents但方法其实有状态于是SynchronizedSubscriber那层锁被绕过了多个线程池线程同时进同一个实例。反之如果你用的是默认同步分发锁天然不需要问题就被掩盖了。解决先确认监听器是不是无状态有状态就老实去掉AllowConcurrentEvents让 Guava 给你加锁。但要注意synchronized (this)锁的粒度是整个实例实例内多个方法互斥如果方法耗时长会拖慢所有订阅者这时候正确做法是把状态挪到方法外而不是靠锁硬扛。5.4 给Object挂了一个订阅者然后所有事件都往这里灌现象某个「通用日志」监听器订阅了Object结果所有事件都进去一次包括框架内部的、你不关心的日志量暴涨。原因3.2 的flattenHierarchy会把Object展开进每一个事件类型的层级里所以订阅Object等价于全量订阅。这不是 bug是设计。解决通用处理用DeadEvent或者显式定义一个DomainEvent基类让业务事件都继承它然后订阅基类而不是Object。另外注意如果同时订阅了Object和具体类型同一个事件会命中两次顺序按类型展开顺序来这点在写「兜底处理」时特别容易忘。5.5 注册了没反注册内存缓慢上涨现象长跑的服务堆内存一直涨dump 下来发现大量监听器实例被CopyOnWriteArraySet的数组引用着。原因Subscriber里持有target你的监听器实例的强引用注册表又持有Subscriber。只要 EventBus 的生命周期比监听器长——比如 EventBus 是单例监听器是每次请求 new 出来的——这个监听器就永远回收不掉。异步场景下更明显因为线程池的队列里还可能压着事件的引用。解决短生命周期的监听器用完就unregister用try/finally包住。生命周期长的场景优先做成单例监听器注册一次而不是每次 new。这类问题不会立刻炸都是压测或者跑几天之后才露头属于典型的后悔药没处买。6. 二次开发照着自己的需求重写一个轻量总线读源码最终要落到改。如果只是想在项目里用直接用 Guava 就好真正的需求往往是「我想砍掉依赖」「我想加个拦截器」「我想让事件支持返回值」。这时候照着 EventBus 的结构自己写一份最小实现比在它外面包一层更省事。6.1 三个必须自己决定的改造点第一是注解的保留策略和扫描范围。EventBus 用Retention(RUNTIME)加getDeclaredMethods逐层向上扫好处是能覆盖父类和接口默认方法坏处是每次注册都要走泛型反射。我的做法是注册阶段扫一遍、结果缓存缓存键用监听器Class加弱引用避免类加载器泄漏。第二是注册中心的数据结构。事件类型到处理器的映射我一般用ConcurrentHashMapClass?, CopyOnWriteArrayListHandler保留「读多写少」的假设但不追求和 Guava 完全一致的去重语义——同一个监听器注册两次要不要去重取决于你的业务我倾向于允许重复并打印警告。第三是分发器的形态。同步版本直接把 3.3 的队列逻辑抄过来就行关键是那两个ThreadLocal队列负责拍平递归标记负责界定最外层循环。异步版本我一般不用全局并发队列而是每个事件类型一个Executor这样慢的订阅者不会拖累快的。下面是最小可用的骨架能跑通「注册 投递 父类型匹配」三条Retention(RetentionPolicy.RUNTIME) Target(ElementType.METHOD) public interface OnEvent {} public final class MiniBus { // 事件类型 - 处理器列表读多写少写时复制 private final MapClass?, CopyOnWriteArrayListHandler registry new ConcurrentHashMap(); public void register(Object listener) { // getMethods() 只返回 public且包含继承来的方法天然过滤掉 private 的坑 for (Method m : listener.getClass().getMethods()) { if (!m.isAnnotationPresent(OnEvent.class)) continue; if (m.getParameterCount() ! 1) { throw new IllegalArgumentException(OnEvent 方法必须只有一个参数: m); } registry.computeIfAbsent(m.getParameterTypes()[0], k - new CopyOnWriteArrayList()) .add(new Handler(listener, m)); } } public void post(Object event) { for (Class? type : typeChain(event.getClass())) { // 自己 - 父类 - 接口 for (Handler h : registry.getOrDefault(type, EMPTY)) { h.invoke(event); // 内部捕异常并上报绝不外抛 } } } }typeChain要按「自己、父类链、所有接口」展开并去重顺序和TypeToken的展开保持一致否则迁移时行为会变。Handler.invoke里必须setAccessible(true)之外还要把InvocationTargetException剥壳——这一步偷懒后面所有排查都会痛苦。另外注意getMethods()拿不到父类里非 public 的方法如果你需要支持包级可见的订阅者得换成沿继承链getDeclaredMethods。6.2 怎么验证你的版本和 Guava 行为一致不要靠肉眼比对。用同一组用例分别跑 Guava 和自己的实现对比四个维度回调顺序、异常是否外抛、无人订阅时的行为、父类型匹配的范围。4.5 那三条断言可以直接复用再加一条assertEquals(guavaHits, myHits)。验证通过的标志是把MiniBus替换掉EventBus业务代码一行不改、测试全绿。但凡需要改业务代码说明你的语义已经偏了。从那以后我每次动这类基础设施的源码都强制自己先写一组「行为快照」测试再改——不是测功能是测顺序、异常、边界这些看不见的约定。因为顺序这东西改的时候觉得无关紧要出问题的时候就是黑匣子。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
