消息總線EventBus源碼分析以及與Otto框架對比

(一).前言:

上一篇我們對EventBus的簡介和基本使用做了說明，今天我們主要深入的使用EventBus，同時會從源碼的角度對于訂閱和發送消息做分析，以及和另外的消息總線框架Otto在性能等方面做一個對比分析。

FastDev4Android框架項目地址: https://github.com/jiangqqlmj/FastDev4Android

(二).框架簡單說明:

通過上一篇文章的介紹，EventBus的使用步驟如下:

•   定義一個事件，用于EventBus的分發。
•   定義訂閱者，把該訂閱者加入到EventBus中。
•   通過EventBus.post來進行分發事件,告訴訂閱者有事情發生了。訂閱者接收到信息進行相應處理。
•   使用完成之后，訂閱者需要反注冊取消訂閱。

具體原理圖如下：

訂閱者接收到通知的時候會調用相應的函數進行處理事件，在EventBus中一般有以下四種方法來讓我們進行處理:

1. onEvent
2. onEventMainThread
3. onEventBackground
4. onEventAsync

這四個訂閱方法有很多的相似之處，但是功能上面還是有點不同的，EventBus會通過調用post方法來進行分發消息，讓訂閱者進行接收，訂閱者接收到事件消息是通過上面幾個方法來進行接收和處理的。下面我們來對這四個方法的具體使用場景做一個介紹:

• onEvent：使用該方法作為訂閱函數表示post消息事件和接收消息事件在同一個線程中。
• onEventMainThread:  該方法會在UI  Main線程中運行，接收事件同時會在UI線程中運行，這樣我們就可以在該方法中直接更新UI
• onEventBackground:使用該方法，如果事件是在UI Main線程發出來，該方法會在子線程中執行，如果事件是從子線程中發出來，該onEventBackground方法會在子線程中執行。
• onEventAsync:使用該方法，會在創建新的子線程中執行onEventAsync

/**
* 收到消息 進行相關處理
* @param event
*/
public void onEventMainThread(TestEventFirst event) {
Log.d("zttjiangqq","onEventMainThread收到消息:"+event.getMsg());
textView_one.setText(event.getMsg());
//showToastMsgShort(event.getMsg());
}
/**
* 收到消息 進行相關處理
* @param event
*/
public void onEventMainThread(TestEventSecond event) {
Log.d("zttjiangqq","onEventMainThread收到消息:"+event.getMsg());
textView_two.setText(event.getMsg());
//showToastMsgShort(event.getMsg());
}
/**
* 收到消息 進行相關處理
* @param event
*/
public void onEventMainThread(TestEventThird event) {
Log.d("zttjiangqq","onEventMainThread收到消息:"+event.getMsg());
textView_third.setText(event.getMsg());
//showToastMsgShort(event.getMsg());
}

3.2.演示效果如下:

(四).源碼解析:

以上主要為EventBus的主要使用，現在開始我們對于EventBus的注冊和發送兩個模塊從源碼的角度來走一下。

4.1.EventBus對象獲取:我們一般使用單例模式獲取。保證對象唯一性。

/**
* 采用單例模式獲取EventBus實例對象  一般我們獲取EventBus對象 就是采用這種方式，不建議直接new
* @return
*/
public static EventBus getDefault() {
if (defaultInstance == null) {
synchronized (EventBus.class) {
if (defaultInstance == null) {
defaultInstance = new EventBus();
}
}
}
return defaultInstance;
}

4.2.訂閱模塊:入口，進行訂閱注冊

public void register(Object subscriber) {
register(subscriber, false, 0);
}

•   subscriber:需要注冊的訂閱者,
•   sticky:是否為粘性，這邊默認為false,
•   priority:事件的優先級,默認為0

下面我們來具體看一下register(subscriber, false, 0)方法具體實現的功能:

private synchronized void register(Object subscriber, boolean sticky, int priority) {
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriber.getClass());
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod, sticky, priority);
}
}

該函數中會通過findSubscriberMethods()來獲取所有訂閱的方法，具體主要的步驟我這邊已經進行注釋了

/**
* 進行查找訂閱者中所有訂閱的方法
* @param subscriberClass
* @return  所有訂閱的方法的集合
*/
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
String key = subscriberClass.getName();
List<SubscriberMethod> subscriberMethods;
//從緩存中獲取訂閱的方法，第一次使用肯定緩存中不存在
synchronized (methodCache) {
subscriberMethods = methodCache.get(key);
}
if (subscriberMethods != null) {
return subscriberMethods;
}
//訂閱方法的集合
subscriberMethods = new ArrayList<SubscriberMethod>();
Class<?> clazz = subscriberClass;
HashMap<String, Class> eventTypesFound = new HashMap<String, Class>();
StringBuilder methodKeyBuilder = new StringBuilder();
while (clazz != null) {
String name = clazz.getName();
if (name.startsWith("java.") || name.startsWith("javax.") || name.startsWith("android.")) {
// Skip system classes, this just degrades performance
// 這邊直接跳過了系統類，因為系統類中 普通開發者不會使用在系統類中使用EventBus，所以就忽略處理了，不然會降低性能
break;
}
// Starting with EventBus 2.2 we enforced methods to be public (might change with annotations again)
try {
// This is faster than getMethods, especially when subscribers a fat classes like Activities
// 通過反射來獲取當前類中的所有方法
Method[] methods = clazz.getDeclaredMethods();
// 正式開始查詢所有訂閱的方法
filterSubscriberMethods(subscriberMethods, eventTypesFound, methodKeyBuilder, methods);
} catch (Throwable th) {
th.printStackTrace();
// Workaround for java.lang.NoClassDefFoundError, see https://github.com/greenrobot/EventBus/issues/149 Method[] methods = subscriberClass.getMethods();
subscriberMethods.clear();
eventTypesFound.clear();
filterSubscriberMethods(subscriberMethods, eventTypesFound, methodKeyBuilder, methods);
break;
}
clazz = clazz.getSuperclass();
}
//拋出異常，訂閱者沒有實現onEvent開頭的公共方法
if (subscriberMethods.isEmpty()) {
throw new EventBusException("Subscriber " + subscriberClass + " has no public methods called "
+ ON_EVENT_METHOD_NAME);
} else {
//訂閱的方法存在，同時加入緩存
synchronized (methodCache) {
methodCache.put(key, subscriberMethods);
}
return subscriberMethods;
}
}

然后調用filterSubscriberMethods()進行過濾，把訂閱方法加入到集合中

/**
* 查詢訂閱的方法，查到方法，方法加入到subScriberMethods
* @param subscriberMethods
* @param eventTypesFound
* @param methodKeyBuilder
* @param methods
*/
private void filterSubscriberMethods(List<SubscriberMethod> subscriberMethods,
HashMap<String, Class> eventTypesFound, StringBuilder methodKeyBuilder,
Method[] methods) {
//遍歷類中的所有方法
for (Method method : methods) {
String methodName = method.getName();
//過濾onEvent開頭的方法
if (methodName.startsWith(ON_EVENT_METHOD_NAME)) {
//返回方法修飾符 例如 public,private,protected
int modifiers = method.getModifiers();
Class<?> methodClass = method.getDeclaringClass();
if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
//訂閱方法必須為public類型
//進行獲取方法的參數類型
Class<?>[] parameterTypes = method.getParameterTypes();
if (parameterTypes.length == 1) {
//進行獲取線程模式類型
ThreadMode threadMode = getThreadMode(methodClass, method, methodName);
if (threadMode == null) {
continue;
}
//取出當前傳入的訂閱者
Class<?> eventType = parameterTypes[0];
//methodKeyBuilder key ="0"."methodName".">"."eventType_Name"
methodKeyBuilder.setLength(0);
methodKeyBuilder.append(methodName);
methodKeyBuilder.append('>').append(eventType.getName());
String methodKey = methodKeyBuilder.toString();
Class methodClassOld = eventTypesFound.put(methodKey, methodClass);
if (methodClassOld == null || methodClassOld.isAssignableFrom(methodClass)) {
// Only add if not already found in a sub class
// 構建一個訂閱方法的對象(里面存入方法名，線程模式類型,事件類型)，加入到訂閱方法集合中
subscriberMethods.add(new SubscriberMethod(method, threadMode, eventType));
} else {
// Revert the put, old class is further down the class hierarchy
eventTypesFound.put(methodKey, methodClassOld);
}
}
} else if (!skipMethodVerificationForClasses.containsKey(methodClass)) {
Log.d(EventBus.TAG, "Skipping method (not public, static or abstract): " + methodClass + "."
+ methodName);
}
}
}
}

上面已經進行獲取了所有的訂閱函數，那么現在開始就可以進行訂閱了，讓我們來查看subscribe()方法做的功能操作:

/**
* 開始進行為訂閱者 注冊相關的訂閱方法
* @param subscriber  訂閱者
* @param subscriberMethod  訂閱的方法
* @param sticky    是否為粘性
* @param priority  優先級
*/
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod, boolean sticky, int priority) {
//通過訂閱方法中進行獲取訂閱方法的類型
Class<?> eventType = subscriberMethod.eventType;
//通過訂閱事件的類型 進行獲取所有的訂閱信息(有訂閱者對象，訂閱方法,優先級)
CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
//進行創建一個訂閱者
Subscription newSubscription = new Subscription(subscriber, subscriberMethod, priority);
if (subscriptions == null) {
//如果當前的事件類型不存在訂閱信息，那么就創建一個訂閱信息集合
subscriptions = new CopyOnWriteArrayList<Subscription>();
//同時把當前的訂閱信息加入到該訂閱中
subscriptionsByEventType.put(eventType, subscriptions);
} else {
if (subscriptions.contains(newSubscription)) {
//拋出異常，該訂閱者已經注冊過該事件類中
throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
+ eventType);
}
}
// Starting with EventBus 2.2 we enforced methods to be public (might change with annotations again)
// subscriberMethod.method.setAccessible(true);
// 優先級判斷，進行排序
int size = subscriptions.size();
for (int i = 0; i <= size; i++) {
if (i == size || newSubscription.priority > subscriptions.get(i).priority) {
subscriptions.add(i, newSubscription);
break;
}
}
//將當前的事件加入到訂閱者列表中
List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
if (subscribedEvents == null) {
subscribedEvents = new ArrayList<Class<?>>();
typesBySubscriber.put(subscriber, subscribedEvents);
}
subscribedEvents.add(eventType);
//是否粘性判斷
if (sticky) {
if (eventInheritance) {
// Existing sticky events of all subclasses of eventType have to be considered.
// Note: Iterating over all events may be inefficient with lots of sticky events,
// thus data structure should be changed to allow a more efficient lookup
// (e.g. an additional map storing sub classes of super classes: Class -> List<Class>).
Set<Map.Entry<Class<?>, Object>> entries = stickyEvents.entrySet();
for (Map.Entry<Class<?>, Object> entry : entries) {
Class<?> candidateEventType = entry.getKey();
if (eventType.isAssignableFrom(candidateEventType)) {
Object stickyEvent = entry.getValue();
checkPostStickyEventToSubscription(newSubscription, stickyEvent);
}
}
} else {
Object stickyEvent = stickyEvents.get(eventType);
checkPostStickyEventToSubscription(newSubscription, stickyEvent);
}
}
}

OK完成以上步驟，我們就大體完成了訂閱注冊工作，下面就是需要分析一下post的流程:

主要先看post主函數:

/**
* 向EventBus中發送消息事件對象
* @param event
*/
public void post(Object event) {
PostingThreadState postingState = currentPostingThreadState.get();
//把消息加入到事件隊列中
List<Object> eventQueue = postingState.eventQueue;
eventQueue.add(event);
if (!postingState.isPosting) {
postingState.isMainThread = Looper.getMainLooper() == Looper.myLooper();
postingState.isPosting = true;
if (postingState.canceled) {
throw new EventBusException("Internal error. Abort state was not reset");
}
try {
//當消息隊列不為空的時候，進行這正式發送消息，采用循環，把隊列中所有的消息發送出去
while (!eventQueue.isEmpty()) {
postSingleEvent(eventQueue.remove(0), postingState);
}
} finally {
postingState.isPosting = false;
postingState.isMainThread = false;
}
}
}

然后進行發送功能，調用postSingleEvent()函數方法:

//消息發送:發送單個事件消息
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
if (eventInheritance) {
List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
int countTypes = eventTypes.size();
for (int h = 0; h < countTypes; h++) {
Class<?> clazz = eventTypes.get(h);
subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
}
} else {
subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
}
if (!subscriptionFound) {
if (logNoSubscriberMessages) {
Log.d(TAG, "No subscribers registered for event " + eventClass);
}
if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
eventClass != SubscriberExceptionEvent.class) {
post(new NoSubscriberEvent(this, event));
}
}
}

接著進行消息過濾postSingleEventForEventType()方法

/**
* 進行該特定的Event發送相應的消息
* @param event  事件消息
* @param postingState
* @param eventClass
* @return
*/
private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
CopyOnWriteArrayList<Subscription> subscriptions;
synchronized (this) {
subscriptions = subscriptionsByEventType.get(eventClass);
}
if (subscriptions != null && !subscriptions.isEmpty()) {
for (Subscription subscription : subscriptions) {
postingState.event = event;
postingState.subscription = subscription;
boolean aborted = false;
try {
//發生消息給訂閱者
postToSubscription(subscription, event, postingState.isMainThread);
aborted = postingState.canceled;
} finally {
postingState.event = null;
postingState.subscription = null;
postingState.canceled = false;
}
if (aborted) {
break;
}
}
return true;
}
return false;
}

最終這邊有一個核心的方法:postToSubscription()來進行post消息

/**
* 進行發送消息，同時根據發送過來的線程類型類型，發送消息給特定的訂閱方法來進行執行
* @param subscription  訂閱者
* @param event   執行事件
* @param isMainThread 是否為主線程
*/
private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
switch (subscription.subscriberMethod.threadMode) {
case PostThread:
//直接在本線程中調用訂閱函數
invokeSubscriber(subscription, event);
break;
case MainThread:
if (isMainThread) {
//如果是主線程，直接調用訂閱函數
invokeSubscriber(subscription, event);
} else {
//如果不是主線程，通過handler進行處理
mainThreadPoster.enqueue(subscription, event);
}
break;
case BackgroundThread:
if (isMainThread) {
//如果是主線程，采用runnable 中調用
backgroundPoster.enqueue(subscription, event);
} else {
//子線程，直接調用
invokeSubscriber(subscription, event);
}
break;
case Async:
//加入到子線程中進行發送
asyncPoster.enqueue(subscription, event);
break;
default:
throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
}
}

OK,到這邊基本上完成EventBus的register和post的流程的講解，關于這個核心類EventBus的注釋過的類文件已經上傳了大家可以通過該地址進行下載: EventBus注釋過的類文件

(五).和Otto消息總線框架對比:

Otto是Android中另外一個消息總線庫,它其實是EventBus的變種，該和EventBus有一些相同的方法(例如:register,post,unregister…)，但是這兩者之間也有一些不同之處如下:

除了以上功能不同以外，這邊還有性能上面的差異。為了測試性能問題，我們clone當前EventBus項目的時候，會發現有一個EventBusPerformance項目，我們可以使用的不同場景來比較。

基于下表結果顯示，每一個測試方面 EventBus的性能都大于Otto