原文鏈接：http://blog.csdn.net/u012810020/article/details/70056134
簡述：
在項目中，我們大多數開發(fā)者可能都使用過EventBus,即使沒有使用過但我可以確定Android開發(fā)者也聽說過這個牛X的庫，從誕生到目前EventBus已經更新到3.X版本，可見生命力極強呀。那么這篇博文就從EventBus3.0源碼的角度分析一下其內部處理流程。
使用流程：
- 注冊：
EventBus.getDefault().register(obj)
- 訂閱(消息接收)：
@Subscribe
public void receive(Object event){
}
- 發(fā)布消息：
EventBus.getDefault().post(event)
- 注銷:
EventBus.getDefault().unregister(obj)
源碼分析：
注冊：
EventBus.getDefault().register(obj)
這段代碼做了兩件事情：① EventBus.getDefault() 創(chuàng)建EventBus對象；② register(obj) 方法為obj該類對象注冊EventBus。 那這兩個方法究竟在EventBus中究竟做了哪些工作呢？我們打開EventBus的源碼看一下:
1、EventBus.getDefault()
源碼如下：
public static EventBus getDefault() {
if (defaultInstance == null) {
synchronized (EventBus.class) {
if (defaultInstance == null) {
defaultInstance = new EventBus();
}
}
}
return defaultInstance;
}
看到了吧，EventBus采用單例模式創(chuàng)建EventBus對象，接下來它在構造方法中又做了什么事情呢？
public EventBus() {
this(DEFAULT_BUILDER);
}
在構造方法中其調用了有參構造方法：EventBus(EventBusBuilder builder )，我們再跟進去看一看：
EventBus(EventBusBuilder builder) {
subscriptionsByEventType = new HashMap<>();
typesBySubscriber = new HashMap<>();
stickyEvents = new ConcurrentHashMap<>();
mainThreadPoster = new HandlerPoster(this, Looper.getMainLooper(), 10);
backgroundPoster = new BackgroundPoster(this);
asyncPoster = new AsyncPoster(this);
indexCount = builder.subscriberInfoIndexes != null ? builder.subscriberInfoIndexes.size() : 0;
//默認情況下參數為（null，false，false）
subscriberMethodFinder = new SubscriberMethodFinder(builder.subscriberInfoIndexes,
builder.strictMethodVerification, builder.ignoreGeneratedIndex);
logSubscriberExceptions = builder.logSubscriberExceptions;
logNoSubscriberMessages = builder.logNoSubscriberMessages;
sendSubscriberExceptionEvent = builder.sendSubscriberExceptionEvent;
sendNoSubscriberEvent = builder.sendNoSubscriberEvent;
throwSubscriberException = builder.throwSubscriberException;
eventInheritance = builder.eventInheritance;
executorService = builder.executorService;
}
這段代碼對一些變量進行了初始化，現在就挑重要的變量解釋一下。首先，初始化了3個Map，這3個Map有什么用呢？我們再來一一詳細說一下：
① Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType：key:事件類型（如：String類型或者自定義的事件類型），value:該事件的訂閱者的list集合。當發(fā)送event消息的時候，都是去這里找對應的訂閱者。
② Map<Object, List<Class<?>>> typesBySubscriber：key:事件的訂閱者（如XXXActivity），value：事件類型的運行時類的list集合。當register()和unregister()的時候都是操作這個Map。
③ Map<Class<?>, Object> stickyEvents：維護的是粘性事件的集合，粘性事件也就是當event發(fā)送出去之后再注冊粘性事件的話，該粘性事件也能接收到之前發(fā)送出去的event消息。
其次，初始化3個消息發(fā)送器如下：
mainThreadPoster ：該類繼承自Handler，而且在EventBus中mainThreadPoster屬于主線程Handler，這是因為mainThreadPoster就是為處理“消息接收方法在主線程而消息發(fā)送在子線程 ”這個問題而設計的，所以子線程向主線程發(fā)送消息必須使用主線程的Handler。mainThreadPoster繼承Handler也是為了效率考慮的。
backgroundPoster：該類繼承自Runnable，重寫了run()方法。在run（）方法中將子線程中的消息通過EventBus發(fā)送到主線程。所以這個消息發(fā)送器作用就是處理“消息接收方法在子線程接而消息的發(fā)布在主線程”這樣的問題。
asyncPoster：該類繼承自Runnable，也重寫了run()方法，不過就像名字一樣“異步”，也就是說不管訂閱者是不是在主線程，消息接收方法都會另外開啟一個線程處理消息。
然后，一個重要的初始化對象為subscriberMethodFinder，這個對象利用反射的方法查找每一個接收消息者的方法（也即是添加了“@Subscribe ”注解的方法）。
最后就是一些對EventBusBuilder的一些配置信息。其中eventInheritance和executorService在接下來分析源碼時會經常碰到：① eventInheritance表示我們自定義的待發(fā)布消息事件是否允許繼承,，默認情況下eventInheritance==true。它的作用就是處理業(yè)務時后來新增加業(yè)務后不必再修改代碼，只需要繼承就OK啦（這也符合程序的“開閉原則”）如下：
public class MessageEvent {
public final String message;
public MessageEvent(String message) {
this.message = message;
}
}
public class SubMessageEvent extends MessageEvent {
public SubMessageEvent(String message) {
super(message);
}
}
②executorService：這個線程池是給backgroundPoster和asyncPoster用來處理消息發(fā)送的。這樣做也能夠提高消息發(fā)送的效率。
2、注冊register(Object subscriber )
EventBus的初始化工作已經完畢，我們繼續(xù)看一下EventBus是怎么進行注冊的，在注冊過程中又搞了哪些事情？
public void register(Object subscriber) {
Class<?> subscriberClass = subscriber.getClass();
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder
.findSubscriberMethods(subscriberClass);
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}
在該方法中首先取得注冊者的運行時類對象，拿到運行時類對象后通過注冊者注冊方法查找器SubscriberMethodFinder利用反射的方法找到注冊者類中所有的接收消息的方法，也即是所有添加了注解“Subscribe ”的方法。最后進行通過方法subscribe(subscriber, subscriberMethod)為每一個接收消息的方法進行注冊。流程大致就是這樣的，首先 我們先看一下findSubscriberMethods這個方法：
/**
* 方法描述：獲取該運行時類的所有@Subscribe注解的所有方法
*
* @param subscriberClass @Subscribe注解所屬類的運行時類對象
* @return 注冊EventBus類中@Subscribe注解的所有方法的List集合
*/
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
//先從緩存中查找是否存在消息接收的方法
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}
if (ignoreGeneratedIndex) {
//使用反射方法拿到訂閱者中的訂閱方法
subscriberMethods = findUsingReflection(subscriberClass);
} else {
//使用apt處理器拿到訂閱者中的訂閱方法
subscriberMethods = findUsingInfo(subscriberClass);
}
//如果該消息事件沒有被訂閱則拋出異常
if (subscriberMethods.isEmpty()) {
throw new EventBusException("Subscriber " + subscriberClass
+ " and its super classes have no public methods with the @Subscribe annotation");
} else {
//將查到的方法放到緩存中，以便下次使用
METHOD_CACHE.put(subscriberClass, subscriberMethods);
return subscriberMethods;
}
}
這個方法很重要，在初次使用EventBus3.0的時候也容易出錯的一個點（3.0新增注解）：就是在訂閱事件的方法上沒有添加@Subscribe 注解，所以會碰到下面這個異常：
Caused by: org.greenrobot.eventbus.EventBusException: Subscriber class XXX and its super classes
have no public methods with the @Subscribe annotation
說到這我們還是沒有最終看到EventBus是怎么進行注冊的，OK，回過頭來我們繼續(xù)看注冊：
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
Class<?> eventType = subscriberMethod.eventType;
Log.e(TAG, eventType.getSimpleName());
//將訂閱類Object對象subscriber封裝為EventBus的訂閱類Subscription
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions == null) {
subscriptions = new CopyOnWriteArrayList<>();
subscriptionsByEventType.put(eventType, subscriptions);
} else {
if (subscriptions.contains(newSubscription)) {
throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
+ eventType);
}
}
/**
* 方法描述：對CopyOnWriteArrayList中的Subscription根據優(yōu)先級的高低重新進行排序
*/
int size = subscriptions.size();
for (int i = 0; i <= size; i++) {
if (i == size || subscriberMethod.priority > subscriptions.get(i).subscriberMethod.priority) {
subscriptions.add(i, newSubscription);
break;
}
}
/**
* 方法描述：將開發(fā)者注冊EventBus的類的運行時類添加到subscribedEvents中，并且把該運行時類添加到
* typesBySubscriber中
*/
List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
Log.e(TAG, "typesBySubscriber的");
if (subscribedEvents == null) {
subscribedEvents = new ArrayList<>();
typesBySubscriber.put(subscriber, subscribedEvents);
}
subscribedEvents.add(eventType);
if (subscriberMethod.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);
}
}
}
上面的這段代碼雖然很多，但主要做了幾件事情：① 將注冊的訂閱者封裝為新的Subscription類 ②將訂閱者存儲到Map集合subscriptionsByEventType當中 ③對消息事件接收者根據優(yōu)先級進行重排序 ④添加粘性消息事件
發(fā)布消息：
我們已經分析完了EventBus的注冊過程，接下來我們再來分析一下EventBus的事件發(fā)送過程。
EventBus.getDefault().post(event);
那么這段代碼是如何實現消息的發(fā)送呢？繼續(xù)源碼看一下：
public void post(Object event) {
PostingThreadState postingState = currentPostingThreadState.get();
List<Object> eventQueue = postingState.eventQueue;
eventQueue.add(event);//將該事件添加到事件隊列當中
//事件沒有分發(fā)則開始分發(fā)
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 {
//循環(huán)發(fā)送消息
while (!eventQueue.isEmpty()) {
postSingleEvent(eventQueue.remove(0), postingState);
}
} finally {
postingState.isPosting = false;
postingState.isMainThread = false;
}
}
}
從上面的代碼中可以得知，待發(fā)送的消息首先存儲到一個消息list集合當中，然后再不斷的循環(huán)發(fā)送消息。發(fā)送消息時利用的方法是postSingleEvent（Object event, PostingThreadState postingState ），OK，我們繼續(xù)跟進：
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
//默認情況下Event事件允許繼承，即默認情況下eventInheritance==true
if (eventInheritance) {
//查找event事件及event子類事件
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) {
...
}
}
在這個方法中并沒有真正的看到消息的分發(fā)，而是查找了待分發(fā)事件消息及其子類或者是待分發(fā)消息接口及其子類的所有事件（默認情況下我們定義的消息事件是允許繼承的。 我們在項目中起初可能考慮的不是很全面，再到后來不可預料的需求到來時我們可能會繼續(xù)改事件的一種情況，看到這不得不說EventBus真心考慮周全呀）。然后調用postSingleEventForEventType(event, postingState, eventClass)方法查找該事件及其子類事件的訂閱者，如果沒有找到就發(fā)送空消息并打印日志。好吧，很失望，到現在依然沒有看到對消息事件進行分發(fā)。那我們繼續(xù)跟進：postSingleEventForEventType(event, postingState, eventClass)；
private boolean postSingleEventForEventType(Object event, PostingThreadState postingState,
Class<?> eventClass) {
CopyOnWriteArrayList<Subscription> subscriptions;
synchronized (this) {
//從Map中取出所有訂閱了eventClass事件的所有訂閱者
subscriptions = subscriptionsByEventType.get(eventClass);
}
//如果該事件的訂閱者存在則向每一個訂閱者發(fā)布消息事件
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;
}
好吧，小眼一瞄仍然沒有對消息進行分發(fā)，而是查找事件的所有訂閱者然后對所有訂閱者進行了一層封裝，封裝成PostingThreadState。那我們還是繼續(xù)吧，我們跟進postToSubscription(subscription, event, postingState.isMainThread)這個方法：
private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
switch (subscription.subscriberMethod.threadMode) {
case POSTING:
invokeSubscriber(subscription, event);
break;
case MAIN:
if (isMainThread) {
invokeSubscriber(subscription, event);
} else {
mainThreadPoster.enqueue(subscription, event);
}
break;
case BACKGROUND:
if (isMainThread) {
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);
}
}
看到這是不是有一種“復行數十步，豁然開朗”的感覺。是的，在這個方法中我們終于看到了對消息事件進行4中不同情況下的分發(fā)了。根據消息接收的threadMode分別進行了不同的處理：
POSTING：EventBus默認情況下的threadMode類型，這里意思就是如果消息發(fā)布和消息接收在同一線程情況下就直接調用invokeSubscriber(subscription, event)對消息進行發(fā)送。這種情況下事件傳遞是同步完成的，事件傳遞完成時，所有的訂閱者將已經被調用一次了。這個ThreadMode意味著最小的開銷，因為它完全避免了線程的切換。
MAIN：消息接收在主線程中進行（此種情況適合進行UI操作），如果消息發(fā)布也在主線程就直接調用invokeSubscriber(subscription, event)對消息進行發(fā)送（這種情況是POSTING的情況），如果消息發(fā)布不在主線程中進行，那么調用mainThreadPoster.enqueue(subscription, event)進行處理。他是怎么處理的呢？我們跟進去瞧瞧:
final class HandlerPoster extends Handler {
private final PendingPostQueue queue;
private final int maxMillisInsideHandleMessage;
private final EventBus eventBus;
private boolean handlerActive;
//默認情況下EventBus創(chuàng)建HandlerPoster的Looper為MainLooper，最大maxMillisInsideHandleMessage==10ms
HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
super(looper);
this.eventBus = eventBus;
this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
queue = new PendingPostQueue();
}
/**
* 方法描述：將訂閱者與消息實體之間的映射存到隊列PendingPostQueue當中
*
* @param subscription 訂閱者
* @param event 消息事件
*/
void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
synchronized (this) {
queue.enqueue(pendingPost);
if (!handlerActive) {
handlerActive = true;
if (!sendMessage(obtainMessage())) {
throw new EventBusException("Could not send handler message");
}
}
}
}
@Override
public void handleMessage(Message msg) {
boolean rescheduled = false;
try {
long started = SystemClock.uptimeMillis();
while (true) {
//同步取出消息隊列
PendingPost pendingPost = queue.poll();
if (pendingPost == null) {
synchronized (this) {
// Check again, this time in synchronized
pendingPost = queue.poll();
if (pendingPost == null) {
handlerActive = false;
return;
}
}
}
eventBus.invokeSubscriber(pendingPost);
long timeInMethod = SystemClock.uptimeMillis() - started;
//消息發(fā)送超時
if (timeInMethod >= maxMillisInsideHandleMessage) {
if (!sendMessage(obtainMessage())) {
throw new EventBusException("Could not send handler message");
}
rescheduled = true;
return;
}
}
} finally {
handlerActive = rescheduled;
}
}
}
為了說明問題，我們整個類貼出來：由于是主線程向子線程發(fā)送消息所以Looper采用的是主線程Looper，Handler也就是主線程Handler，其內部維護了一個PendingPost的對象池，這樣做也是為了提高內存利用率，這也不是重點，我們直接看重點，在enqueue(Subscription subscription, Object event)方法中利用HandlerPoster 發(fā)送空消息，HandlerPoster也重寫了handleMessage方法，在handleMessage方法中又調用eventBus.invokeSubscriber(pendingPost)進行消息發(fā)送，我們跟進去之后發(fā)現最終還是調用了invokeSubscriber(subscription, event)對消息進行發(fā)送。
BACKGROUND：這種情況是消息接收在子線程（此種模式下適合在接收者方法中做IO等耗時操作）。那么如果消息發(fā)布也在某個子線程中進行的就直接調用invokeSubscriber(subscription, event)對消息進行發(fā)送，如果消息發(fā)布在主線程當中應該盡可能快的將消息發(fā)送出去以免造成主線程阻塞，所以這時候就交給backgroundPoster去處理。它是怎么處理的呢？我們進去看一看：
final class BackgroundPoster implements Runnable {
....
private volatile boolean executorRunning;
BackgroundPoster(EventBus eventBus) {
this.eventBus = eventBus;
queue = new PendingPostQueue();
}
public void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
synchronized (this) {
...
eventBus.getExecutorService().execute(this);
}
}
}
@Override
public void run() {
try {
while (true) {
...
eventBus.invokeSubscriber(pendingPost);
}
} finally {
executorRunning = false;
}
}
}
backgroundPoster對Runnable進行了重寫，而且和HandlerPoster一樣也采用了對象池提高效率，當然重點是其開啟了線程池處理消息的發(fā)送，這也是避免阻塞主線程的舉措。當然其最終還是調用了invokeSubscriber（）-----》invokeSubscriber(subscription, event)方法。
ASYNC：這種情況是消息接收在子線程（如果消息發(fā)布在子線程中進行，那么該子線程既不同于消息發(fā)布的子線程，又不在主線程，而是接收消息是一個獨立于主線程又不同于消息發(fā)布的子線程）。由于在這種模式下每一個新添加的任務都會在線程池中開辟一個新線程執(zhí)行，所以并發(fā)量更高效。而且最終還是會調用invokeSubscriber(subscription, event)方法對消息進行分發(fā)。
既然4種模式下均是調用了invokeSubscriber(subscription, event)方法，那我們最后再看一下這個方法：
void invokeSubscriber(Subscription subscription, Object event) {
try {
subscription.subscriberMethod.method.invoke(subscription.subscriber, event);
} catch (InvocationTargetException e) {
handleSubscriberException(subscription, event, e.getCause());
} catch (IllegalAccessException e) {
throw new IllegalStateException("Unexpected exception", e);
}
}
看到了吧，這個方法中就是利用反射將消息發(fā)送給每一個消息的訂閱者。到此我們就完整的看完了EventBus的工作流程及主要代碼的分析過程。真心不容易呀，已經被累跪啦！
