EventBus

EventBus總結 

1. 采用訂閱者模式（觀察者模式）設計的簡化Activity、Fragment、threads、Service等之間通信的一種事件框架機制，
讓代碼簡潔，耦合性更低。
2. 主要核心內容：注冊（事件訂閱）函數、事件發布函數、反注冊（取消訂閱）函數。

EventBus 需要解決的問題

在日常編碼里面，我們會遇到很多網絡請求，數據庫操作等等，一般情況下這些操作都是通過觀察者模式來實現的。通過Volley來簡單舉個例子：

ImageRequest request = new ImageRequest(url, new Response.Listener<Bitmap>() { @Override public void onResponse(Bitmap bitmap) { mImageView.setImageBitmap(bitmap); } }, 0, 0, null, new Response.ErrorListener() { public void onErrorResponse(VolleyError error) { mImageView.setImageResource(R.drawable.image_load_error); } });

 

此時，你會發現並且開始思考一個問題，如果很多觀察者模式需要使用了？ 比如，你正在開發一個東西，需要監聽網絡狀態變化，App的安裝情況，內容的下載情況。當存在很多觀察者模式，「如何將這些事件通知到監聽者」是可以復用的模塊，這就是EventBus存在的意義。這里需要大家想明白一個問題，觀察者模式本身就是一個可以復用的模塊。

• 內容下載模塊
• 電量監聽模塊
• App按照通知

他們都可以通過EventBus將自身的事件發布出去，使用者只需要在這個模塊里面，注冊對於自己感興趣的內容就行。

EventBus 帶來的好處和引入的問題

好處比較明顯，就是獨立出一個發布訂閱模塊，調用者可以通過使用這個模塊，屏蔽一些線程切換問題，簡單地實現發布訂閱功能。

壞處可能比較隱晦，但這些需要足夠引起我們的重視

• 大量的濫用，將導致邏輯的分散，出現問題后很難定位。
• 沒辦法實現強類型，在編譯的時候就發現問題，（Otto實現了這個，但性能有問題）。在實現上通過一個很弱的協議，比如onEvent{XXX}, {XXX}表示ThreadModel，來實現線程的切換。后面在代碼解析的時候，會說明這個問題。
• 代碼可讀性有些問題，IDE無法識別這些協議，對IDE不友好。

總得來說，如果項目里面有大量的事件交互，那么還是可以通過EventBus來實現，否則還是推薦自己在模塊內部實現觀察者模式

EventBus 源碼解析

EventBus.java

源碼閱讀從外觀類開始，這里是 EventBus.java，核心接口都在這個類里面實現，對內容感興趣的調用方使用 register 方法，當有事件產生的時候，會在onEvent的時候收到相應的回調。

register(Object object);

registerSticky(Object object);

unRegister(Object object);

post(Object object);

 

先看看初始化部分，看看如何實現單例的（可選的）。

// volatile 這里是需要重視的，這個關鍵字保證了defaultInstance在不同線程間的可見性，也就是說在多線程環境下，看到的仍然是最新修改的值。 static volatile EventBus defaultInstance; /** Convenience singleton for apps using a process-wide EventBus instance. */ public static EventBus getDefault() { // 這一步不存在線程問題，volatile保證了。如果沒有defaultInstance實例化出來， if (defaultInstance == null) { synchronized (EventBus.class) { // 進入同步塊的時候，不能保證defaultInstance沒有被實例化出來，所以需要進行double-check if (defaultInstance == null) { defaultInstance = new EventBus(); } } } return defaultInstance; } // 這里實現的時候，考慮的是defaultInstance 不一定是每個人都需要創建的，否則沒必要使用lazy的實現方式 // 下面是一種實現方式 static { defaultInstance = new EventBus(); }

 

EventBus實現了EventBusBuilder，通過Builder的方式使得構建的時候更加容易

public static EventBusBuilder builder() { return new EventBusBuilder(); }

 

下面重點看看register(Object subscriber, boolean sticky, int priority)方法

private synchronized void register(Object subscriber, boolean sticky, int priority) { // 用 subscriberMethodFinder 提供的方法，找到在 subscriber 這個類里面，訂閱的內容。 List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriber.getClass()); for (SubscriberMethod subscriberMethod : subscriberMethods) { // 遍歷這些方法，subscribe 這些事件 subscribe(subscriber, subscriberMethod, sticky, priority); } }

 

findSubscriberMethods 這個方法是實現 EventBus 的核心代碼，這里面包含了 EventBus 隱式定義的交互協議。從這個方法里面，可以看到如何爭取地使用EventBus。

List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
    String key = subscriberClass.getName();
    List<SubscriberMethod> subscriberMethods;
    // 如果這個 Class 對應的方法被緩存，直接返回。 synchronized (methodCache) { subscriberMethods = methodCache.get(key); } // 這個方法其實可以放在 前面的 synchronized 模塊里面 if (subscriberMethods != null) { return subscriberMethods; } subscriberMethods = new ArrayList<SubscriberMethod>(); Class<?> clazz = subscriberClass; HashSet<String> eventTypesFound = new HashSet<String>(); StringBuilder methodKeyBuilder = new StringBuilder(); while (clazz != null) { String name = clazz.getName(); // 跳過JDK里面的類 if (name.startsWith("java.") || name.startsWith("javax.") || name.startsWith("android.")) { // Skip system classes, this just degrades performance break; } // Starting with EventBus 2.2 we enforced methods to be public (might change with annotations again) // 獲取所有聲明的方法 Method[] methods = clazz.getDeclaredMethods(); for (Method method : methods) { String methodName = method.getName(); if (methodName.startsWith(ON_EVENT_METHOD_NAME)) { int modifiers = method.getModifiers(); if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) { // 方法是 Public 的 Class<?>[] parameterTypes = method.getParameterTypes(); if (parameterTypes.length == 1) { String modifierString = methodName.substring(ON_EVENT_METHOD_NAME.length()); ThreadMode threadMode; // 方法的前綴是否是 ‘OnEvent’， 如果是‘OnEvent’，查看后面的字符串，這里定義了 4 種基本類型 // ThreadModel 會在后面介紹 if (modifierString.length() == 0) { threadMode = ThreadMode.PostThread; } else if (modifierString.equals("MainThread")) { threadMode = ThreadMode.MainThread; } else if (modifierString.equals("BackgroundThread")) { threadMode = ThreadMode.BackgroundThread; } else if (modifierString.equals("Async")) { threadMode = ThreadMode.Async; } else { if (skipMethodVerificationForClasses.containsKey(clazz)) { continue; } else { throw new EventBusException("Illegal onEvent method, check for typos: " + method); } } // 獲取參數類型 Class<?> eventType = parameterTypes[0]; methodKeyBuilder.setLength(0); methodKeyBuilder.append(methodName); methodKeyBuilder.append('>').append(eventType.getName()); // 得到類似於一個句柄的東西，比如 onEventMainThread>DownloadInfo String methodKey = methodKeyBuilder.toString(); if (eventTypesFound.add(methodKey)) { // Only add if not already found in a sub class subscriberMethods.add(new SubscriberMethod(method, threadMode, eventType)); } } } else if (!skipMethodVerificationForClasses.containsKey(clazz)) { Log.d(EventBus.TAG, "Skipping method (not public, static or abstract): " + clazz + "." + methodName); } } } // 這里要為 EventBus 點個贊了，EventBus 是支持繼承的 clazz = clazz.getSuperclass(); } 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; } }

 

現在看下如何把 subscriberClass 里面的內容訂閱到 EventBus 里面去。

// Must be called in synchronized block private void subscribe(Object subscriber, SubscriberMethod subscriberMethod, boolean sticky, int priority) { Class<?> eventType = subscriberMethod.eventType; // 獲取訂閱了某種類型數據的 Subscription 。 使用了 CopyOnWriteArrayList ，這個是線程安全的， // CopyOnWriteArrayList 會在更新的時候，重新生成一份 copy，其他線程使用的是 // copy，不存在什么線程安全性的問題。 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) { // 是否支持繼承，這個可以在 Builder 的時候指定，如果不支持，那么可能有20%以上的性能提升 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 { // 檢查是否有 sticky 的event， 如果存在就發布出去 Object stickyEvent = stickyEvents.get(eventType); checkPostStickyEventToSubscription(newSubscription, stickyEvent); } } }

 

// 取消訂閱的代碼 public synchronized void unregister(Object subscriber) { List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber); if (subscribedTypes != null) { for (Class<?> eventType : subscribedTypes) { unubscribeByEventType(subscriber, eventType); } typesBySubscriber.remove(subscriber); } else { Log.w(TAG, "Subscriber to unregister was not registered before: " + subscriber.getClass()); } }

 

// 重點看看發布的方法 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)); } } }

 

EventBus ThreadModel

EventBus 一共提供了 4 種 ThreadModel， 分別是 PostThread, MainThread, BackgroundThread, 
Async。

• PostThread 默認實現，執行發生在同一個線程
• MainThread 執行在UI 線程上
• BackgroundThread 回調發生在非 UI 線程上
• Async 永遠執行在一個其他的線程上

以上這四種類型，足以支持觀察者模式里面需要進行的異步處理。

EventBus 如何實現線程轉換的

但凡經歷一些實際項目，就會發現，經常存在「生產」和「消費」沖突的情況，這里就需要使用「生產者與消費者」模式。

EventBus 中 生產者和消費者模式的實現主要是在 PendingPostQueue里面。

PendingPostQueue 的實現比較簡單，主要是通過在 enqueue 和 poll 的時候進行 synchronized 同步來實現的。

synchronized void enqueue(PendingPost pendingPost) { if (pendingPost == null) { throw new NullPointerException("null cannot be enqueued"); } // 將 Post 插入到隊列尾部 if (tail != null) { tail.next = pendingPost; tail = pendingPost; } else if (head == null) { // 在最開始的時候，建立頭部和尾部的索引 head = tail = pendingPost; } else { throw new IllegalStateException("Head present, but no tail"); } notifyAll(); } synchronized PendingPost poll() { PendingPost pendingPost = head; // 從頭部獲取 if (head != null) { head = head.next; if (head == null) { tail = null; } } return pendingPost; }

 

// 這里需要注意的地方是 PendingPost， 這里維護了一個 pendingPostPool 的池子， 當PendingPost 不再需要的時候，就釋放回池子里面去，避免了新建對象的開銷。 static void releasePendingPost(PendingPost pendingPost) { pendingPost.event = null; pendingPost.subscription = null; pendingPost.next = null; synchronized (pendingPostPool) { // Don't let the pool grow indefinitely if (pendingPostPool.size() < 10000) { pendingPostPool.add(pendingPost); } } }

 

EventBus 如何發布時間的

// 每個ThreadModel (除了PostThread) 都維護了一個 Poster， 這個Post 里面維持了一個 ``生產者消費者模式``， 來消費和使用事件。 private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) { switch (subscription.subscriberMethod.threadMode) { case PostThread: invokeSubscriber(subscription, event); break; case MainThread: // 主線程的poster if (isMainThread) { invokeSubscriber(subscription, event); } else { mainThreadPoster.enqueue(subscription, event); } break; case BackgroundThread: 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); } }