Guava是Google开源的一个Java基础类库,它在Google内部被广泛使用。Guava提供了很多功能模块比如:集合、并发库、缓存等,EventBus是其中的一个module,本篇结合EventBus源码来谈谈它的设计与实现。

概要

首先,我们先来预览一下EventBus模块的全部类图:

类并不是多而且几乎没有太多继承关系。

下面,我们来看一下各个类的职责:

  • EventBus:核心类,代表了一个事件总线。Publish事件也由它发起。
  • AsyncEventBus:在分发事件的时候,将其压入一个全局队列的异步分发模式。
  • Subscriber:对某个事件的处理器抽象,封装了事件的订阅者以及处理器,并负责事件处理(该类的类名及其语义有些不明确,后续会谈到)。
  • SubscriberRegistry:订阅注册表,它用于存储Subscriber跟Event的对应关系,以便于EventBus在publish一个事件时,可以找到它对应的Subscriber。
  • Dispatcher:事件分发器,它定义了事件的分发策略。
  • @Subscribe:用于标识事件处理器的注解,当EventBus publish一个事件后,相应的Subscriber将会得到通知并执行事件处理器。
  • @AllowConcurrentEvents:该注解跟@Subscribe一同使用,标识该订阅者的处理方法为线程安全的,该注解还用于标识该方法将可能会被EventBus在多线程环境下执行。
  • DeadEvent:死信(没有订阅者关注的事件)对象。
  • SubscribeExceptionHandler:订阅者抛出异常的处理器。
  • SubscribeExceptionContext:订阅者抛出异常的上下文对象。
在对每个类进行分解之前,我们再来看一下各个类之间的关联关系:

分“类”解读

EventBus

它有这么几个字段:
  • identifier:事件总线的标识,这说明在一个应用里是可以有多个EventBus的。如果不指明它的值,它将以“default”作为其默认名称。
  • executor:它是Executor接口的实例,用于对订阅者处理事件方法的执行。这里需要注意的是,该字段的实例化是在EventBus内部构造器中,并不是从外部注入进来的,另外真正的执行订阅者方法的时机也不由EventBus负责,而是由Subscriber负责,因此该字段会被公开给外部访问。
  • exceptionHandler:它是SubscribeExceptionHandler的实例,用于处理订阅者在执行事件处理方法时抛出的异常。EventBus可以接收一个外部定义的异常处理器,也可以采用内部缺省的日志记录处理器。
  • subscribers:订阅者注册表,用于存储所有的事件以及事件处理器、订阅对象的对应关系。
  • dispatcher:事件分发器,用于分发事件给订阅对象的事件处理器,该对象在EventBus构造方法内部初始化,默认的实现是PerThreadQueuedDispatcher,该分发器将事件存入队列,并保证在同一个线程上发送的事件能够按照他们发布的顺序被分发给所有的订阅者。
EventBus提供了几个核心方法:

  • register:注册subscriber;
  • unregister:移除注册过的subscriber;
  • post:发布事件;

你可以将EventBus看做是一个代理,这些方法真正的实现者都是上面的这些对象。

AsyncEventBus

一个支持异步发布模式的EventBus,它覆盖了EventBus的默认构造方法,指定了一个异步的分发器:LegacyAsyncDispatcher,这个分发器基于一个全局的队列来暂存未发布的事件。

Subscriber

之前也提到Subscriber的名称是比较容易混淆的。这个类的名称看似表示一个订阅者对象,但其实是用来封装“一个订阅者的一个事件处理器”对象。因为当一个订阅者存在多个处理方法被标注为@Subscribe的时候,那么每个处理方法都对应于一个独立的Subscriber对象的实例。我个人觉得这个名称与其具体的实现语义有些混淆。当然也许实现者认为:一个对象以及一个事件处理器就是一个Subscriber的话,那是没有问题的。因此这里为了理解方便,你可以将其看做是一个封装了订阅者对象以及一个订阅者处理器方法的实体类。
Subscriber的访问级别是package的,它还承担了执行事件处理的责任。通过一个create静态工厂方法创建它:
static Subscriber create(EventBus bus, Object listener, Method method) {return isDeclaredThreadSafe(method)? new Subscriber(bus, listener, method): new SynchronizedSubscriber(bus, listener, method);}

它接收三个参数:

  • bus:EventBus的实例,通过它来获取事件的执行器(executor)
  • listener:真实的订阅者对象
  • method:订阅对象的事件处理方法的Method实例
在实现中,它会先判断该处理器方法上是否被标注有@AllowConcurrentEvents注解,如果有,则实例化Subscriber类的一个实例;如果没有,则不允许eventbus在多线程的环境中调用处理器方法,所以这里专门为此提供了一个同步的订阅者对象:SynchronizedSubscriber来保证线程安全。

该类的两个关键方法之一:
dispatchEvent:
final void dispatchEvent(final Object event) {executor.execute(new Runnable() {@Overridepublic void run() {try {invokeSubscriberMethod(event);} catch (InvocationTargetException e) {bus.handleSubscriberException(e.getCause(), context(event));}}});}

它调用一个多线程执行器来执行事件处理器方法。

另一个方法:invokeSubscriberMethod以反射的方式调用事件处理器方法。
另外,该类对Object的equals方法进行了override并标识为final。主要是为了避免同一个对象对某个事件进行重复订阅,在SubscriberRegistry中会有相应的判等操作。当然这里Subscriber也override并final了hashCode方法。这是最佳实践,不必多谈,如果不了解的可以去看看《Effective Java》。
该类还有个内部类,就是我们上面谈到的SynchronizedSubscriber,它继承了Subscriber,与Subscriber唯一的不同就是在invokeSubscriberMethod的执行上做了同步。

SubscriberRegistry

针对单个EventBus的订阅与事件的关系维护。在内部用来存储订阅者关系的对象是java并发包下的并发Map:ConcurrentMap,该map以Class对象为键,值的类型是CopyOnWriteArraySet<Subscriber>集合类型。
SubscriberRegistry直接依赖EventBus对象,所以在构造器中需要注入EventBus的实例。
SubscriberRegistry里有两个关键的实例方法:register/unregister。

register

接收订阅者对象作为参数并建立Event跟Subscriber的关联关系。
我们来看看它的实现:
void register(Object listener) {Multimap<Class<?>, Subscriber> listenerMethods = findAllSubscribers(listener);for (Map.Entry<Class<?>, Collection<Subscriber>> entry : listenerMethods.asMap().entrySet()) {Class<?> eventType = entry.getKey();Collection<Subscriber> eventMethodsInListener = entry.getValue();CopyOnWriteArraySet<Subscriber> eventSubscribers = subscribers.get(eventType);if (eventSubscribers == null) {CopyOnWriteArraySet<Subscriber> newSet = new CopyOnWriteArraySet<Subscriber>();eventSubscribers = MoreObjects.firstNonNull(subscribers.putIfAbsent(eventType, newSet), newSet);}eventSubscribers.addAll(eventMethodsInListener);}}

它首先获得一个Multimap实例(它是Google Guava集合框架提供的一个多值Map类型,也就是说一个key可以对应多个value),该Multimap用于存储事件类型对应的该订阅者内所有关于该事件的处理器方法集合,其key为事件的Class类型。这里在for循环的中通过asMap获取其map视图,即可将Multimap对应的多个值存储到一个Collection中。

也就是说这里for循环的每个entry,表示的是一个事件的Class实例对应的一组Subscriber的集合,即eventMethodsInListener。
然后根据该事件的Class对象从注册表中获取对应的存储Subscriber实例的集合,如果不存在则创建该集合,然后将该订阅者内所有的事件处理器方法都加入到注册表中去。

unregister

unregister的实现跟register有些类似,先查找该订阅者所有的事件类型与处理器的对应关系。然后,遍历所有的事件类型,移除针对当前订阅者的所有Subscriber实例。

findAllSubscribers

register/unregister方法都调用了findAllSubscribers方法,它有一些特别之处,这里需要单独拎出来提一下。
findAllSubscribers用于查找事件类型以及事件处理器的对应关系。查找注解需要涉及到反射,通过反射来获取标注在方法上的注解。因为Guava针对EventBus的注册采取的是“隐式契约”而非接口这种“显式契约”。而类与接口是存在继承关系的,所有很有可能某个订阅者其父类(或者父类实现的某个接口)也订阅了某个事件。因此这里的查找需要顺着继承链向上查找父类的方法是否也被注解标注,代码实现:
  private Multimap<Class<?>, Subscriber> findAllSubscribers(Object listener) {Multimap<Class<?>, Subscriber> methodsInListener = HashMultimap.create();Class<?> clazz = listener.getClass();for (Method method : getAnnotatedMethods(clazz)) {Class<?>[] parameterTypes = method.getParameterTypes();Class<?> eventType = parameterTypes[0];methodsInListener.put(eventType, Subscriber.create(bus, listener, method));}return methodsInListener;}

同样涉及这个问题的,还有根据事件类型获取Subscriber实例的方法:getSubscribers。

getSubscribers

  Iterator<Subscriber> getSubscribers(Object event) {ImmutableSet<Class<?>> eventTypes = flattenHierarchy(event.getClass());List<Iterator<Subscriber>> subscriberIterators =Lists.newArrayListWithCapacity(eventTypes.size());for (Class<?> eventType : eventTypes) {CopyOnWriteArraySet<Subscriber> eventSubscribers = subscribers.get(eventType);if (eventSubscribers != null) {// eager no-copy snapshotsubscriberIterators.add(eventSubscribers.iterator());}}return Iterators.concat(subscriberIterators.iterator());}

Dispatcher

dispatcher用于分发事件给Subscriber。它内部实现了多个分发器用于提供在不同场景下不同的事件顺序性。Dispatcher是一个抽象类,定义了一个核心抽象方法:

abstract void dispatch(Object event, Iterator<Subscriber> subscribers);

该方法用于将一个指定的事件分发给所有的订阅者。

另外在Dispatcher提供了三个不同的分发器实现:

PerThreadQueuedDispatcher

它比较常用,针对每个线程构建一个队列用于暂存事件对象。保证所有的事件都按照他们publish的顺序从单一的线程上发出。保证从单一线程上发出,没什么特别的地方,主要是在内部定义了一个队列,将其放在ThreadLocal中,用以跟特定的线程关联。

LegacyAsyncDispatcher

另一个异步分发器的实现:LegacyAsyncDispatcher,之前在介绍AsyncEventBus的时候提到,它就是用这种实现来分发事件。
它在内部通过一个ConcurrentLinkedQueue<EventWithSubscriber>的全局队列来存储事件。从关键方法:dispatch的实现来看,它跟PerThreadQueuedDispatcher的区别主要是两个循环上的差异(这里基于队列的缓存事件的方式,肯定会存在两个循环:循环取队列里的事件以及循环发送给Subscriber)。
PerThreadQueuedDispatcher:是两层嵌套循环,外层是遍历队列取事件,内存是遍历事件的订阅处理器。
LegacyAsyncDispatcher:是一前一后两个循环。前面一个是遍历事件订阅处理器,并构建一个事件实体对象存入队列。后一个循环是遍历该事件实体对象队列,取出事件实体对象中的事件进行分发。

ImmediateDispatcher

其实以上两个基于中间队列的分发实现都可以看做是异步模式,而ImmediateDispatcher则是同步模式:只要有事件发生就会立即分发并被立即得到处理。ImmediateDispatcher从感官上看类似于线性并顺序执行,而采用队列的方式有多线程汇聚到一个公共队列的由发散到聚合的模型。因此,ImmediateDispatcher的分发方式是一种深度优先的方式,而使用队列是一种广度优先的方式。

DeadEvent

它是一个实体对象,封装了没有订阅者的事件。DeadEvent由两个属性组成:
  • source:事件源(通常指发布事件的EventBus对象)
  • event:事件对象
DeadEvent对象的产生:当通过某个EventBus的实例发布一个事件的时候,没有找到事件订阅者并且它本身又不是一个DeadEvent的实例时,将由EventBus构建一个DeadEvent类的实例。

总结

Guava的EventBus源码还是比较简单、清晰的。从源码来看,它一番常用的Observer的设计方式,放弃采用统一的接口、统一的事件对象类型。转而采用基于注解扫描的绑定方式。
其实无论是强制实现统一的接口,还是基于注解的实现方式都是在构建一种关联关系(或者说满足某种契约)。很明显接口的方式是编译层面上强制的显式契约,而注解的方式则是运行时动态绑定的隐式契约关系。接口的方式是传统的方式,编译时确定观察者关系,清晰明了,但通常要求有一致的事件类型、方法签名。而基于注解实现的机制,刚好相反,编译时因为没有接口的语法层面上的依赖关系,显得不那么清晰,至少静态分析工具很难展示观察者关系,但无需一致的方法签名、事件参数,至于多个订阅者类之间的继承关系,可以继承接收事件的通知,可以看作既是其优点也是其缺点。
原文发布时间为:2015-06-01
本文作者:vinoYang
本文来自云栖社区合作伙伴CSDN博客,了解相关信息可以关注CSDN博客。

Google-Guava-EventBus源码解读相关推荐

  1. Google guava cache源码解析1--构建缓存器(2)

    此文已由作者赵计刚授权网易云社区发布. 欢迎访问网易云社区,了解更多网易技术产品运营经验. CacheBuilder-->maximumSize(long size) /*** 指定cache中 ...

  2. Google guava cache源码解析1--构建缓存器(3)

    此文已由作者赵计刚授权网易云社区发布. 欢迎访问网易云社区,了解更多网易技术产品运营经验. 下面介绍在LocalCache(CacheBuilder, CacheLoader)中调用的一些方法: Ca ...

  3. spark源码解读3之RDD中top源码解读

    更多代码请见:https://github.com/xubo245/SparkLearning spark源码解读系列环境:spark-2.0.1 (20161103github下载版) 1.理解 输 ...

  4. Bert系列(三)——源码解读之Pre-train

    https://www.jianshu.com/p/22e462f01d8c pre-train是迁移学习的基础,虽然Google已经发布了各种预训练好的模型,而且因为资源消耗巨大,自己再预训练也不现 ...

  5. Ubuntu 16.04下Caffe-SSD的应用(四)——ssd_pascal.py源码解读

    前言 caffe-ssd所有的训练时的参数,全部由ssd_pascal.py来定义,之后再去调用相关的脚本和函数,所以想要训练自己的数据,首先要明白ssd_pascal.py各个定义参数的大体意思. ...

  6. Android 开源框架之 Android-async-http 源码解读

    开源项目链接 Android-async-http仓库:https://github.com/loopj/android-async-http android-async-http主页:http:// ...

  7. 源码解读之Pre-train

    pre-train是迁移学习的基础,虽然Google已经发布了各种预训练好的模型,而且因为资源消耗巨大,自己再预训练也不现实(在Google Cloud TPU v2 上训练BERT-Base要花费近 ...

  8. RTC 月度小报 5 月 | WebRTC M83、SOLO 源码解读、实时美声……

    本月亮点速览 RTC开发者社区: CSDN专访RTC编程大赛获奖者 如何高效实现PSTN/SIP接入实时音视频网络 在线教育的创新模式及AI应用实践 开源与技术科普: WebRTC M83 Relea ...

  9. golang源码解读之 net.Dial

    1.调用 例子 conn, err := net.Dial("tcp", "google.com:80") if err != nil {// handle e ...

最新文章

  1. 解析codepage和charset的含义及其应用
  2. Trinity 1.1 发布,Linux 桌面环境
  3. html 子级选择器,css3子选择器(选择某个标签做内容)
  4. [书目20071127]图书 时间陷阱 目录
  5. 中医:看脚十秒钟可知身体疾病
  6. 掌握神经网络,我应该学习哪些至关重要的知识点?
  7. FastJson的使用方法总结
  8. 图像处理中ct图的通道是多少_新一代安检CT机,智能安防领域又一明星产品
  9. getconnection java_在MyEclipse用java写的一个GetConnection1.java,用于连接MySQL,却总是出错。(没有财富值了,见谅!)...
  10. 实用小程序,快速求A类不确定度(物理实验),保留六位
  11. linux学习书籍汇总 值得推荐的linux学习书籍
  12. ModifyStyle()
  13. Internet结构和ISP
  14. 男男java_猛男学习Java的第二天
  15. 使用Visual Paradigm如何复制表格
  16. 揭露安利!!!!!!
  17. web前端数据表格有合并项的一种简单实现方法
  18. 面试问烂了的测试用例: 登录界面的测试用例
  19. 接口中的变量为什么不能是普通变量,只能是static final
  20. 本周大新闻|沙特PIF再投Magic Leap,周融资超5.1亿美元

热门文章

  1. duration java_Java Duration类| minusMinutes()方法与示例
  2. 1补码 2补码_8085微处理器中8位数字的1和2的补码
  3. mui实现分享功能_MUI 分享功能(微信、QQ 、朋友圈)
  4. linux ida 图形界面,linux – IDA在屏幕内不起作用
  5. Ubuntu14.04设置开机可以以root用户身份登录
  6. c语言 包络算法,包络检测C程序
  7. java毛玻璃_模糊效果(毛玻璃效果)
  8. 操作系统(四)文件管理
  9. django-celery定时任务以及异步任务and服务器部署并且运行全部过程
  10. 去除list集合中重复项的几种方法