EventBus是Guava的事件處理機(jī)制，是設(shè)計(jì)模式中的觀察者模式（生產(chǎn)/消費(fèi)者編程模型）的優(yōu)雅實(shí)現(xiàn)落蝙。對(duì)于事件監(jiān)聽(tīng)和發(fā)布訂閱模式，EventBus是一個(gè)非常優(yōu)雅和簡(jiǎn)單解決方案，我們不用創(chuàng)建復(fù)雜的類和接口層次結(jié)構(gòu)赐纱。
demo
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ù)類型觸發(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類型的參數(shù)的方法將會(huì)在觸發(fā)事件的時(shí)候被觸發(fā)瓜客。
從上面的使用中，我們可以看出竿开，EventBus與觀察者模式不同的地方在于：當(dāng)注冊(cè)了一個(gè)監(jiān)聽(tīng)者的時(shí)候谱仪，只有當(dāng)某個(gè)方法使用了@Subscribe注解聲明并且參數(shù)與發(fā)布的事件類型匹配，那么這個(gè)方法才會(huì)被觸發(fā)否彩。這就是說(shuō)疯攒，同一個(gè)監(jiān)聽(tīng)者可以監(jiān)聽(tīng)多種類型的事件，也可以在多次監(jiān)聽(tīng)同一個(gè)事件列荔。
源碼分析
下面我們來(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)康姆椒ǜ哦袛嘣摲椒ㄊ欠袷褂昧薂Subscribe并且方法的參數(shù)類型是否與post()方法發(fā)布的事件類型一致呼巷，如果一致的話，那么我們就使用反射來(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)題的？
EventBus事件總線
- register：把監(jiān)聽(tīng)器中申明的所有訂閱事件方法注冊(cè)到SubscriberRegistry（訂閱者注冊(cè)器）中胶惰。
- post：發(fā)布事件給所有已注冊(cè)過(guò)的訂閱者傻工，最終開(kāi)啟線程完成訂閱方法。
@Beta
public class EventBus {
private final String identifier;//事件總線標(biāo)識(shí)：用于自定義標(biāo)識(shí)這個(gè)事件總線
private final Executor executor;//默認(rèn)的線程執(zhí)行器孵滞，用于把事件轉(zhuǎn)發(fā)給訂閱者
private final SubscriberRegistry subscribers = new SubscriberRegistry(this);//訂閱注冊(cè)器
private final Dispatcher dispatcher;//事件轉(zhuǎn)發(fā)器
15 //構(gòu)造器：使用默認(rèn)字符串
public EventBus() {
this("default");
}
//構(gòu)造器：使用自定義字符串
public EventBus(String identifier) {
this(
identifier,
MoreExecutors.directExecutor(),
Dispatcher.perThreadDispatchQueue(),
LoggingHandler.INSTANCE);
} 58
//注冊(cè)監(jiān)聽(tīng)者中申明的所有訂閱方法（@Subscribe標(biāo)記的）中捆，用以接收事件
public void register(Object object) {
subscribers.register(object);
}
// 解除訂閱
public void unregister(Object object) {
subscribers.unregister(object);
}
//發(fā)布事件給所有已注冊(cè)過(guò)的訂閱者
public void post(Object event) {
// 找到事件的所有訂閱者
Iterator<Subscriber> eventSubscribers = subscribers.getSubscribers(event);
if (eventSubscribers.hasNext()) {
// 事件轉(zhuǎn)發(fā)器，把事件轉(zhuǎn)發(fā)給訂閱者
dispatcher.dispatch(event, eventSubscribers);
} else if (!(event instanceof DeadEvent)) {
// 如果該事件即沒(méi)有訂閱者坊饶，也沒(méi)事DeadEvent,那么封裝成DeadEvent并重新發(fā)布
post(new DeadEvent(this, event));
}
...省略非重要方法167 }
- subscribers是SubscriberRegistry類型的泄伪，實(shí)際上EventBus在添加、移除和遍歷觀察者的時(shí)候都會(huì)使用該實(shí)例的方法匿级，所有的觀察者信息也都維護(hù)在該實(shí)例中.
- executor是事件分發(fā)過(guò)程中使用到的線程池蟋滴，可以自己實(shí)現(xiàn)染厅； dispatcher是Dispatcher類型的子類，用來(lái)在發(fā)布事件的時(shí)候分發(fā)消息給監(jiān)聽(tīng)者津函，它有幾個(gè)默認(rèn)的實(shí)現(xiàn)肖粮，分別針對(duì)不同的分發(fā)方式；
- exceptionHandler是SubscriberExceptionHandler類型的尔苦，它用來(lái)處理異常信息涩馆，在默認(rèn)的EventBus實(shí)現(xiàn)中，會(huì)在出現(xiàn)異常的時(shí)候打印出log蕉堰，當(dāng)然我們也可以定義自己的異常處理策咯凌净。
如果我們想要了解EventBus是如何注冊(cè)和取消注冊(cè)以及如何遍歷來(lái)觸發(fā)事件的，就應(yīng)該從SubscriberRegistry入手.
我們需要在EventBus中維護(hù)幾個(gè)映射屋讶，以便在發(fā)布事件的時(shí)候找到并通知所有的監(jiān)聽(tīng)者冰寻，首先是事件類型->觀察者列表的映射。
EventBus中發(fā)布事件是針對(duì)各個(gè)方法的皿渗，我們將一個(gè)事件對(duì)應(yīng)的類型信息和方法信息等都維護(hù)在一個(gè)對(duì)象中斩芭，在EventBus中就是觀察者Subscriber. 然后，通過(guò)事件類型映射到觀察者列表乐疆，當(dāng)發(fā)布事件的時(shí)候划乖，只要根據(jù)事件類型到列表中尋找所有的觀察者并觸發(fā)監(jiān)聽(tīng)方法即可。 在SubscriberRegistry中通過(guò)如下數(shù)據(jù)結(jié)構(gòu)來(lái)完成這一映射：
private final ConcurrentMap<Class<?>, CopyOnWriteArraySet<Subscriber>> subscribers = Maps.newConcurrentMap();
從上面的定義形式中我們可以看出挤土，這里使用的是事件的Class類型映射到Subscriber列表的琴庵。這里的Subscriber列表使用的是Java中的CopyOnWriteArraySet集合，
它底層使用了CopyOnWriteArrayList仰美，并對(duì)其進(jìn)行了封裝迷殿，也就是在基本的集合上面增加了去重的操作。這是一種適用于讀多寫少場(chǎng)景的集合咖杂，在讀取數(shù)據(jù)的時(shí)候不會(huì)加鎖庆寺，
寫入數(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)者的類型
Class<?> clazz = listener.getClass();
// 獲取上述監(jiān)聽(tīng)者的全部監(jiān)聽(tīng)方法
UnmodifiableIterator var4 = getAnnotatedMethods(clazz).iterator(); // 1
// 遍歷上述方法墨叛，并且根據(jù)方法和類型參數(shù)創(chuàng)建觀察者并將其插入到映射表中
while(var4.hasNext()) {
Method method = (Method)var4.next();
Class<?>[] parameterTypes = method.getParameterTypes();
// 事件類型
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ǔ)事件類型到觀察者的一對(duì)多映射忍疾。
- 調(diào)用SubscriberRegistry的register(listener)來(lái)執(zhí)行注冊(cè)監(jiān)聽(tīng)器。
- register步驟如下：
EventBus-包含-》SubscriberRegistry-包含-》ConcurrentMap<Class<?>, CopyOnWriteArraySet<Subscriber>> subscribers 用以維護(hù)事件和訂閱者的映射
注意下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ù)雜翩活。
private static ImmutableList<Method> getAnnotatedMethodsNotCached(Class<?> clazz) {
//獲取超類class集合
Set<? extends Class<?>> supertypes = TypeToken.of(clazz).getTypes().rawTypes();
Map<MethodIdentifier, Method> identifiers = Maps.newHashMap();
//遍歷超類
for (Class<?> supertype : supertypes) {
//遍歷超類中的所有定義的方法
for (Method method : supertype.getDeclaredMethods()) {
//如果方法上有@Subscribe注解
if (method.isAnnotationPresent(Subscribe.class) && !method.isSynthetic()) {
// 方法的參數(shù)類型數(shù)組
Class<?>[] parameterTypes = method.getParameterTypes();
// 校驗(yàn)：事件訂閱方法必須只能有一個(gè)參數(shù)阱洪，即事件類
checkArgument(
parameterTypes.length == 1,
"Method %s has @Subscribe annotation but has %s parameters."
+ "Subscriber methods must have exactly 1 parameter.",
method,
parameterTypes.length);
// 封裝方法定義對(duì)象
MethodIdentifier ident = new MethodIdentifier(method);
// 去重并添加進(jìn)map
if (!identifiers.containsKey(ident)) {
identifiers.put(ident, method);
}
}
}
}
// map轉(zhuǎn)ImmutableList
return ImmutableList.copyOf(identifiers.values());
}
這樣，我們就分析完了findAllSubscribers()方法菠镇，整理一下：當(dāng)注冊(cè)監(jiān)聽(tīng)者的時(shí)候冗荸，首先會(huì)拿到該監(jiān)聽(tīng)者的類型，然后從緩存中嘗試獲取該監(jiān)聽(tīng)者對(duì)應(yīng)的所有監(jiān)聽(tīng)方法利耍，如果沒(méi)有的話就遍歷該類的方法進(jìn)行獲取蚌本，并添加到緩存中；
然后隘梨，會(huì)遍歷上述拿到的方法集合程癌，根據(jù)事件的類型（從方法參數(shù)得知）和監(jiān)聽(tīng)者等信息創(chuàng)建一個(gè)觀察者，并將事件類型-觀察者鍵值對(duì)插入到一個(gè)一對(duì)多映射表中并返回出嘹。
下面席楚，我們看下EventBus中的register()方法的代碼：
void register(Object listener) {
// 獲取事件類型-觀察者映射表
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()方法類似咬崔，我們不進(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));
}
}
當(dāng)調(diào)用EventBus.post()方法的時(shí)候回先用SubscriberRegistry的getSubscribers方法獲取該事件對(duì)應(yīng)的全部觀察者
Iterator<Subscriber> getSubscribers(Object event) {
// 獲取事件類型的所有父類型和自身構(gòu)成的集合
ImmutableSet<Class<?>> eventTypes = flattenHierarchy(event.getClass()); // 3
List<Iterator<Subscriber>> subscriberIterators = Lists.newArrayListWithCapacity(eventTypes.size());
UnmodifiableIterator var4 = eventTypes.iterator();
// 遍歷上述事件類型垮斯，并從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());
}
Dispatcher
從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è)字類AsyncEventBus就是用該分發(fā)器進(jìn)行事件分發(fā)的失仁。
- PerThreadQueuedDispatcher：這種分發(fā)器使用了兩個(gè)線程局部變量進(jìn)行控制，當(dāng)dispatch()方法被調(diào)用的時(shí)候们何，會(huì)先獲取當(dāng)前線程的觀察者隊(duì)列萄焦，并將傳入的觀察者列表傳入到該隊(duì)列中；然后通過(guò)一個(gè)布爾類型的線程局部變量冤竹，判斷當(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類型也是通過(guò)EventBus獲取到的片部。（原來(lái)EventBus中的構(gòu)造方法中的字段是在這里用到的Ａ秃狻）
小結(jié)
EventBus中維護(hù)了三個(gè)緩存和四個(gè)映射：
- 事件類型到觀察者列表的映射（緩存）
- 事件類型到監(jiān)聽(tīng)者方法列表的映射（緩存）
- 事件類型到事件類型及其所有父類的類型的列表的映射（緩存）
- 觀察者到監(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)進(jìn)行觸發(fā)
每次使用EventBus注冊(cè)和取消注冊(cè)監(jiān)聽(tīng)者的時(shí)候惰说，都會(huì)先從緩存中進(jìn)行獲取，不是每一次都會(huì)用到反射的缘回，這可以提升獲取的效率吆视，也解答了我們一開(kāi)始提出的效率的問(wèn)題。當(dāng)使用反射觸發(fā)方法的調(diào)用貌似是不可避免的了酥宴。
ventBus中使用了非常多的數(shù)據(jù)結(jié)構(gòu)啦吧，比如MultiMap、CopyOnWriteArraySet等拙寡，還有一些緩存和映射的工具庫(kù)授滓，這些大部分都來(lái)自于Guava。
