EventBus的設(shè)計(jì)理念是基于觀察者模式的，可以參考設(shè)計(jì)模式(1)—觀察者模式先來(lái)了解該設(shè)計(jì)模式。
1、程序示例
EventBus的使用是非常簡(jiǎn)單的，首先你要添加Guava的依賴(lài)到自己的項(xiàng)目中。這里我們通過(guò)一個(gè)最基本的例子來(lái)說(shuō)明EveentBus是如何使用的。
public static void main(String...args) {
// 定義一個(gè)EventBus對(duì)象，這里的Joker是該對(duì)象的id
EventBus eventBus = new EventBus("Joker");
// 向上述EventBus對(duì)象中注冊(cè)一個(gè)監(jiān)聽(tīng)對(duì)象
eventBus.register(new EventListener());
// 使用EventBus發(fā)布一個(gè)事件，該事件會(huì)給通知到所有注冊(cè)的監(jiān)聽(tīng)者
eventBus.post(new Event("Hello every listener, joke begins..."));
}
// 事件，監(jiān)聽(tīng)者監(jiān)聽(tīng)的事件的包裝對(duì)象
public static class Event {
public String message;
Event(String message) {
this.message = message;
}
}
// 監(jiān)聽(tīng)者
public static class EventListener {
// 監(jiān)聽(tīng)的方法，必須使用注解聲明，且只能有一個(gè)參數(shù)，實(shí)際觸發(fā)一個(gè)事件的時(shí)候會(huì)根據(jù)參數(shù)類(lèi)型觸發(fā)方法
@Subscribe
public void listen(Event event) {
System.out.println("Event listener 1 event.message = " + event.message);
}
}
首先，這里我們封裝了一個(gè)事件對(duì)象Event，一個(gè)監(jiān)聽(tīng)者對(duì)象EventListener。然后，我們用EventBus的構(gòu)造方法創(chuàng)建了一個(gè)EventBus實(shí)例，并將上述監(jiān)聽(tīng)者實(shí)例注冊(cè)進(jìn)去。然后，我們使用上述EventBus實(shí)例發(fā)布一個(gè)事件Event。然后，以上注冊(cè)的監(jiān)聽(tīng)者中的使用@Subscribe注解聲明并且只有一個(gè)Event類(lèi)型的參數(shù)的方法將會(huì)在觸發(fā)事件的時(shí)候被觸發(fā)。
總結(jié)：從上面的使用中，我們可以看出，EventBus與觀察者模式不同的地方在于：當(dāng)注冊(cè)了一個(gè)監(jiān)聽(tīng)者的時(shí)候，只有當(dāng)某個(gè)方法使用了@Subscribe注解聲明并且參數(shù)與發(fā)布的事件類(lèi)型匹配，那么這個(gè)方法才會(huì)被觸發(fā)。這就是說(shuō)，同一個(gè)監(jiān)聽(tīng)者可以監(jiān)聽(tīng)多種類(lèi)型的事件，也可以在多次監(jiān)聽(tīng)同一個(gè)事件。
2、EventBus源碼分析
2.1 分析之前
好了，通過(guò)上面的例子，我們了解了EventBus最基本的使用方法。下面我們來(lái)分析一下在Guava中是如何為我們實(shí)現(xiàn)這個(gè)API的。不過(guò)，首先，我們還是先試著考慮一下自己設(shè)計(jì)這個(gè)API的時(shí)候如何設(shè)計(jì)，并且提出幾個(gè)問(wèn)題，然后帶著問(wèn)題到源碼中尋找答案。
假如要我們?nèi)ピO(shè)計(jì)這樣一個(gè)API，最簡(jiǎn)單的方式就是在觀察者模式上進(jìn)行拓展：每次調(diào)用EventBus.post()方法的時(shí)候，會(huì)對(duì)所有的觀察者對(duì)象進(jìn)行遍歷，然后獲取它們?nèi)康姆椒?，判斷該方法是否使用?code>@Subscribe并且方法的參數(shù)類(lèi)型是否與post()方法發(fā)布的事件類(lèi)型一致，如果一致的話，那么我們就使用反射來(lái)觸發(fā)這個(gè)方法。在觀察者模式中，每個(gè)觀察者都要實(shí)現(xiàn)一個(gè)接口，發(fā)布事件的時(shí)候，我們只要調(diào)用接口的方法就行，但是EventBus把這個(gè)限制設(shè)定得更加寬泛，也就是監(jiān)聽(tīng)者無(wú)需實(shí)現(xiàn)任何接口，只要方法使用了注解并且參數(shù)匹配即可。
從上面的分析中可以看出，這里面不僅要對(duì)所有的監(jiān)聽(tīng)者進(jìn)行遍歷，還要對(duì)它們的方法進(jìn)行遍歷，找到了匹配的方法之后又要使用反射來(lái)觸發(fā)這個(gè)方法。首先，當(dāng)注冊(cè)的監(jiān)聽(tīng)者數(shù)量比較多的時(shí)候，鏈?zhǔn)秸{(diào)用的效率就不高；然后我們又要使用反射來(lái)觸發(fā)匹配的方法，這樣效率肯定又低了一些。那么在Guava的EventBus中是如何解決這兩個(gè)問(wèn)題的？
另外還要注意下下文中的觀察者和監(jiān)聽(tīng)者的不同，監(jiān)聽(tīng)者用來(lái)指我們使用EventBus.register()注冊(cè)的對(duì)象，觀察者是EventBus中的對(duì)象Subscriber，后者封裝了一個(gè)監(jiān)聽(tīng)者的所有的信息，比如監(jiān)聽(tīng)的方法等等。
一般我們是不會(huì)直接操作Subscriber對(duì)象的，它的訪問(wèn)權(quán)限也只在EventBus的包中可訪問(wèn)。
2.2 著手分析
首先，當(dāng)我們使用new初始化一個(gè)EventBus的時(shí)候，實(shí)際都會(huì)調(diào)用到下面的這個(gè)方法：
EventBus(String identifier, Executor executor, Dispatcher dispatcher, SubscriberExceptionHandler exceptionHandler) {
this.subscribers = new SubscriberRegistry(this);
this.identifier = (String)Preconditions.checkNotNull(identifier);
this.executor = (Executor)Preconditions.checkNotNull(executor);
this.dispatcher = (Dispatcher)Preconditions.checkNotNull(dispatcher);
this.exceptionHandler = (SubscriberExceptionHandler)Preconditions.checkNotNull(exceptionHandler);
}
這里的identifier是一個(gè)字符串類(lèi)型，類(lèi)似于EventBus的id；
subscribers是SubscriberRegistry類(lèi)型的，實(shí)際上EventBus在添加、移除和遍歷觀察者的時(shí)候都會(huì)使用該實(shí)例的方法，所有的觀察者信息也都維護(hù)在該實(shí)例中；
executor是事件分發(fā)過(guò)程中使用到的線程池，可以自己實(shí)現(xiàn)；
dispatcher是Dispatcher類(lèi)型的子類(lèi)，用來(lái)在發(fā)布事件的時(shí)候分發(fā)消息給監(jiān)聽(tīng)者，它有幾個(gè)默認(rèn)的實(shí)現(xiàn)，分別針對(duì)不同的分發(fā)方式；
exceptionHandler是SubscriberExceptionHandler類(lèi)型的，它用來(lái)處理異常信息，在默認(rèn)的EventBus實(shí)現(xiàn)中，會(huì)在出現(xiàn)異常的時(shí)候打印出log，當(dāng)然我們也可以定義自己的異常處理策咯。
所以，從上面的分析中可以看出，如果我們想要了解EventBus是如何注冊(cè)和取消注冊(cè)以及如何遍歷來(lái)觸發(fā)事件的，就應(yīng)該從SubscriberRegistry入手。確實(shí)，個(gè)人也認(rèn)為，這個(gè)類(lèi)的實(shí)現(xiàn)也是EventBus中最精彩的部分。
2.2.1 SubscriberRegistry
根據(jù)2.1中的分析，我們需要在EventBus中維護(hù)幾個(gè)映射，以便在發(fā)布事件的時(shí)候找到并通知所有的監(jiān)聽(tīng)者，首先是事件類(lèi)型->觀察者列表的映射。
上面我們也說(shuō)過(guò)，EventBus中發(fā)布事件是針對(duì)各個(gè)方法的，我們將一個(gè)事件對(duì)應(yīng)的類(lèi)型信息和方法信息等都維護(hù)在一個(gè)對(duì)象中，在EventBus中就是觀察者Subscriber。
然后，通過(guò)事件類(lèi)型映射到觀察者列表，當(dāng)發(fā)布事件的時(shí)候，只要根據(jù)事件類(lèi)型到列表中尋找所有的觀察者并觸發(fā)監(jiān)聽(tīng)方法即可。
在SubscriberRegistry中通過(guò)如下數(shù)據(jù)結(jié)構(gòu)來(lái)完成這一映射：
private final ConcurrentMap<Class<?>, CopyOnWriteArraySet<Subscriber>> subscribers = Maps.newConcurrentMap();
從上面的定義形式中我們可以看出，這里使用的是事件的Class類(lèi)型映射到Subscriber列表的。這里的Subscriber列表使用的是Java中的CopyOnWriteArraySet集合，
它底層使用了CopyOnWriteArrayList，并對(duì)其進(jìn)行了封裝，也就是在基本的集合上面增加了去重的操作。這是一種適用于讀多寫(xiě)少場(chǎng)景的集合，在讀取數(shù)據(jù)的時(shí)候不會(huì)加鎖，
寫(xiě)入數(shù)據(jù)的時(shí)候進(jìn)行加鎖，并且會(huì)進(jìn)行一次數(shù)組拷貝。
既然，我們已經(jīng)知道了在SubscriberRegistry內(nèi)部會(huì)在注冊(cè)的時(shí)候向以上數(shù)據(jù)結(jié)構(gòu)中插入映射，那么我們可以具體看下它是如何完成這一操作的。
在分析register()方法之前，我們先看下SubscriberRegistry內(nèi)部經(jīng)常使用的幾個(gè)方法，它們的原理與我們上面提出的問(wèn)題息息相關(guān)。
首先是findAllSubscribers()方法，它用來(lái)獲取指定監(jiān)聽(tīng)者對(duì)應(yīng)的全部觀察者集合。下面是它的代碼：
private Multimap<Class<?>, Subscriber> findAllSubscribers(Object listener) {
// 創(chuàng)建一個(gè)哈希表
Multimap<Class<?>, Subscriber> methodsInListener = HashMultimap.create();
// 獲取監(jiān)聽(tīng)者的類(lèi)型
Class<?> clazz = listener.getClass();
// 獲取上述監(jiān)聽(tīng)者的全部監(jiān)聽(tīng)方法
UnmodifiableIterator var4 = getAnnotatedMethods(clazz).iterator(); // 1
// 遍歷上述方法，并且根據(jù)方法和類(lèi)型參數(shù)創(chuàng)建觀察者并將其插入到映射表中
while(var4.hasNext()) {
Method method = (Method)var4.next();
Class<?>[] parameterTypes = method.getParameterTypes();
// 事件類(lèi)型
Class<?> eventType = parameterTypes[0];
methodsInListener.put(eventType, Subscriber.create(this.bus, listener, method));
}
return methodsInListener;
}
這里注意一下Multimap數(shù)據(jù)結(jié)構(gòu)，它是Guava中提供的集合結(jié)構(gòu)，與普通的哈希表不同的地方在于，它可以完成一對(duì)多操作。這里用來(lái)存儲(chǔ)事件類(lèi)型到觀察者的一對(duì)多映射。
注意下1處的代碼，我們上面也提到過(guò)，當(dāng)新注冊(cè)監(jiān)聽(tīng)者的時(shí)候，用反射獲取全部方法并進(jìn)行判斷的過(guò)程非常浪費(fèi)性能，而這里就是這個(gè)問(wèn)題的答案：
這里getAnnotatedMethods()方法會(huì)嘗試從subscriberMethodsCache中獲取所有的注冊(cè)監(jiān)聽(tīng)的方法（即使用了注解并且只有一個(gè)參數(shù)），下面是這個(gè)方法的定義：
private static ImmutableList<Method> getAnnotatedMethods(Class<?> clazz) {
return (ImmutableList)subscriberMethodsCache.getUnchecked(clazz);
}
這里的subscriberMethodsCache的定義是：
private static final LoadingCache<Class<?>, ImmutableList<Method>> subscriberMethodsCache = CacheBuilder.newBuilder().weakKeys().build(new CacheLoader<Class<?>, ImmutableList<Method>>() {
public ImmutableList<Method> load(Class<?> concreteClass) throws Exception { // 2
return SubscriberRegistry.getAnnotatedMethodsNotCached(concreteClass);
}
});
這里的作用機(jī)制是：當(dāng)使用subscriberMethodsCache.getUnchecked(clazz)獲取指定監(jiān)聽(tīng)者中的方法的時(shí)候會(huì)先嘗試從緩存中進(jìn)行獲取，如果緩存中不存在就會(huì)執(zhí)行2處的代碼，
調(diào)用SubscriberRegistry中的getAnnotatedMethodsNotCached()方法獲取這些監(jiān)聽(tīng)方法。這里我們省去該方法的定義，具體可以看下源碼中的定于，其實(shí)就是使用反射并完成一些校驗(yàn)，并不復(fù)雜。
這樣，我們就分析完了findAllSubscribers()方法，整理一下：當(dāng)注冊(cè)監(jiān)聽(tīng)者的時(shí)候，首先會(huì)拿到該監(jiān)聽(tīng)者的類(lèi)型，然后從緩存中嘗試獲取該監(jiān)聽(tīng)者對(duì)應(yīng)的所有監(jiān)聽(tīng)方法，如果沒(méi)有的話就遍歷該類(lèi)的方法進(jìn)行獲取，并添加到緩存中；
然后，會(huì)遍歷上述拿到的方法集合，根據(jù)事件的類(lèi)型（從方法參數(shù)得知）和監(jiān)聽(tīng)者等信息創(chuàng)建一個(gè)觀察者，并將事件類(lèi)型-觀察者鍵值對(duì)插入到一個(gè)一對(duì)多映射表中并返回。
下面，我們看下EventBus中的register()方法的代碼：
void register(Object listener) {
// 獲取事件類(lèi)型-觀察者映射表
Multimap<Class<?>, Subscriber> listenerMethods = this.findAllSubscribers(listener);
Collection eventMethodsInListener;
CopyOnWriteArraySet eventSubscribers;
// 遍歷上述映射表并將新注冊(cè)的觀察者映射表添加到全局的subscribers中
for(Iterator var3 = listenerMethods.asMap().entrySet().iterator(); var3.hasNext(); eventSubscribers.addAll(eventMethodsInListener)) {
Entry<Class<?>, Collection<Subscriber>> entry = (Entry)var3.next();
Class<?> eventType = (Class)entry.getKey();
eventMethodsInListener = (Collection)entry.getValue();
eventSubscribers = (CopyOnWriteArraySet)this.subscribers.get(eventType);
// 如果指定事件對(duì)應(yīng)的觀察者列表不存在就創(chuàng)建一個(gè)新的
if (eventSubscribers == null) {
CopyOnWriteArraySet<Subscriber> newSet = new CopyOnWriteArraySet();
eventSubscribers = (CopyOnWriteArraySet)MoreObjects.firstNonNull(this.subscribers.putIfAbsent(eventType, newSet), newSet);
}
}
}
SubscriberRegistry中的register()方法與unregister()方法類(lèi)似，我們不進(jìn)行說(shuō)明。下面看下當(dāng)調(diào)用EventBus.post()方法的時(shí)候的邏輯。下面是其代碼：
public void post(Object event) {
// 調(diào)用SubscriberRegistry的getSubscribers方法獲取該事件對(duì)應(yīng)的全部觀察者
Iterator<Subscriber> eventSubscribers = this.subscribers.getSubscribers(event);
if (eventSubscribers.hasNext()) {
// 使用Dispatcher對(duì)事件進(jìn)行分發(fā)
this.dispatcher.dispatch(event, eventSubscribers);
} else if (!(event instanceof DeadEvent)) {
this.post(new DeadEvent(this, event));
}
}
從上面的代碼可以看出，實(shí)際上當(dāng)調(diào)用EventBus.post()方法的時(shí)候回先用SubscriberRegistry的getSubscribers方法獲取該事件對(duì)應(yīng)的全部觀察者，所以我們需要先看下這個(gè)邏輯。
以下是該方法的定義：
Iterator<Subscriber> getSubscribers(Object event) {
// 獲取事件類(lèi)型的所有父類(lèi)型和自身構(gòu)成的集合
ImmutableSet<Class<?>> eventTypes = flattenHierarchy(event.getClass()); // 3
List<Iterator<Subscriber>> subscriberIterators = Lists.newArrayListWithCapacity(eventTypes.size());
UnmodifiableIterator var4 = eventTypes.iterator();
// 遍歷上述事件類(lèi)型，并從subscribers中獲取所有的觀察者列表
while(var4.hasNext()) {
Class<?> eventType = (Class)var4.next();
CopyOnWriteArraySet<Subscriber> eventSubscribers = (CopyOnWriteArraySet)this.subscribers.get(eventType);
if (eventSubscribers != null) {
subscriberIterators.add(eventSubscribers.iterator());
}
}
return Iterators.concat(subscriberIterators.iterator());
}
這里注意以下3處的代碼，它用來(lái)獲取當(dāng)前事件的所有的父類(lèi)包含自身的類(lèi)型構(gòu)成的集合，也就是說(shuō)，加入我們觸發(fā)了一個(gè)Interger類(lèi)型的事件，那么Number和Object等類(lèi)型的監(jiān)聽(tīng)方法都能接收到這個(gè)事件并觸發(fā)。這里的邏輯很簡(jiǎn)單，就是根據(jù)事件的類(lèi)型，找到它及其所有的父類(lèi)的類(lèi)型對(duì)應(yīng)的觀察者并返回。
2.2.2 Dispatcher
接下來(lái)我們看真正的分發(fā)事件的邏輯是什么樣的。
從EventBus.post()方法可以看出，當(dāng)我們使用Dispatcher進(jìn)行事件分發(fā)的時(shí)候，需要將當(dāng)前的事件和所有的觀察者作為參數(shù)傳入到方法中。然后，在方法的內(nèi)部進(jìn)行分發(fā)操作。最終某個(gè)監(jiān)聽(tīng)者的監(jiān)聽(tīng)方法是使用反射進(jìn)行觸發(fā)的，這部分邏輯在Subscriber內(nèi)部，而Dispatcher是事件分發(fā)的方式的策略接口。EventBus中提供了3個(gè)默認(rèn)的Dispatcher實(shí)現(xiàn)，分別用于不同場(chǎng)景的事件分發(fā):
-
ImmediateDispatcher：直接在當(dāng)前線程中遍歷所有的觀察者并進(jìn)行事件分發(fā)； -
LegacyAsyncDispatcher：異步方法，存在兩個(gè)循環(huán)，一先一后，前者用于不斷往全局的隊(duì)列中塞入封裝的觀察者對(duì)象，后者用于不斷從隊(duì)列中取出觀察者對(duì)象進(jìn)行事件分發(fā)；實(shí)際上，EventBus有個(gè)字類(lèi)AsyncEventBus就是用該分發(fā)器進(jìn)行事件分發(fā)的。 -
PerThreadQueuedDispatcher：這種分發(fā)器使用了兩個(gè)線程局部變量進(jìn)行控制，當(dāng)dispatch()方法被調(diào)用的時(shí)候，會(huì)先獲取當(dāng)前線程的觀察者隊(duì)列，并將傳入的觀察者列表傳入到該隊(duì)列中；然后通過(guò)一個(gè)布爾類(lèi)型的線程局部變量，判斷當(dāng)前線程是否正在進(jìn)行分發(fā)操作，如果沒(méi)有在進(jìn)行分發(fā)操作，就通過(guò)遍歷上述隊(duì)列進(jìn)行事件分發(fā)。
上述三個(gè)分發(fā)器內(nèi)部最終都會(huì)調(diào)用Subscriber的dispatchEvent()方法進(jìn)行事件分發(fā)：
final void dispatchEvent(final Object event) {
// 使用指定的執(zhí)行器執(zhí)行任務(wù)
this.executor.execute(new Runnable() {
public void run() {
try {
// 使用反射觸發(fā)監(jiān)聽(tīng)方法
Subscriber.this.invokeSubscriberMethod(event);
} catch (InvocationTargetException var2) {
// 使用EventBus內(nèi)部的SubscriberExceptionHandler處理異常
Subscriber.this.bus.handleSubscriberException(var2.getCause(), Subscriber.this.context(event));
}
}
});
}
上述方法中的executor是執(zhí)行器，它是通過(guò)EventBus獲取到的；處理異常的SubscriberExceptionHandler類(lèi)型也是通過(guò)EventBus獲取到的。（原來(lái)EventBus中的構(gòu)造方法中的字段是在這里用到的?。┲劣诜瓷溆|發(fā)方法調(diào)用并沒(méi)有太復(fù)雜的邏輯。
另外還要注意下Subscriber還有一個(gè)字類(lèi)SynchronizedSubscriber，它與一般的Subscriber的不同就在于它的反射觸發(fā)調(diào)用的方法被sychronized關(guān)鍵字修飾，也就是它的觸發(fā)方法是加鎖的、線程安全的。
總結(jié)：
至此，我們已經(jīng)完成了EventBus的源碼分析。簡(jiǎn)單總結(jié)一下：
EventBus中維護(hù)了三個(gè)緩存和四個(gè)映射：
- 事件類(lèi)型到觀察者列表的映射（緩存）；
- 事件類(lèi)型到監(jiān)聽(tīng)者方法列表的映射（緩存）；
- 事件類(lèi)型到事件類(lèi)型及其所有父類(lèi)的類(lèi)型的列表的映射（緩存）；
- 觀察者到監(jiān)聽(tīng)者的映射，觀察者到監(jiān)聽(tīng)方法的映射；
觀察者Subscriber內(nèi)部封裝了監(jiān)聽(tīng)者和監(jiān)聽(tīng)方法，可以直接反射觸發(fā)。而如果是映射到監(jiān)聽(tīng)者的話，還要判斷監(jiān)聽(tīng)者的方法的類(lèi)型來(lái)進(jìn)行觸發(fā)。個(gè)人覺(jué)得這個(gè)設(shè)計(jì)是非常棒的，因?yàn)槲覀儫o(wú)需再在EventBus中維護(hù)一個(gè)映射的緩存了，因?yàn)镾ubscriber中已經(jīng)完成了這個(gè)一對(duì)一的映射。
每次使用EventBus注冊(cè)和取消注冊(cè)監(jiān)聽(tīng)者的時(shí)候，都會(huì)先從緩存中進(jìn)行獲取，不是每一次都會(huì)用到反射的，這可以提升獲取的效率，也解答了我們一開(kāi)始提出的效率的問(wèn)題。當(dāng)使用反射觸發(fā)方法的調(diào)用貌似是不可避免的了。
最后，EventBus中使用了非常多的數(shù)據(jù)結(jié)構(gòu)，比如MultiMap、CopyOnWriteArraySet等，還有一些緩存和映射的工具庫(kù)，這些大部分都來(lái)自于Guava。
看了EventBus的實(shí)現(xiàn)，由衷地感覺(jué)Google的工程師真牛！而Guava中還有許多更加豐富的內(nèi)容值得我們?nèi)ネ诰颍?/p>
了解線程局部遍歷可以參考下我的另一篇博文：ThreadLocal的使用及其源碼實(shí)現(xiàn)
