Android每週一輪子：EventBus

前言

開篇要說聲sorry，限於各類緣由，Okhttp的下篇和OKIO要delay 了，本週先來一個簡單一些的。緩存

EventBus 是一個基於觀察者模式的事件發佈/訂閱框架，開發者能夠經過極少的代碼去實現多個模塊之間的通訊，而不須要以層層傳遞接口的形式去單獨構建通訊橋樑。從而下降因多重回調致使的模塊間強耦合，同時避免產生大量內部類。其能夠很好的應用於Activity之間，Fragment之間，後臺線程之間的通訊，避免使用intent或者handler所帶來的複雜度。其缺點則是可能會形成接口的膨脹。特別是當程序要求大量形式各異的通知，而沒有作出良好的抽象時，代碼中會包含大量的接口，接口數量的增加又會帶來命名、註釋等等一大堆問題。本質上說觀察者要求從零開始實現事件的產生、分發與處理過程，這就要求參與者必須對整個通知過程有着良好的理解。當程序代碼適量時，這是一個合理的要求，然而當程序太大時，這將成爲一種負擔。bash

EventBus基於觀察者模式的Android事件分發總線。markdown

EventBus基本使用

1.定義消息事件MessageEvent，也就是建立事件類型

public class MessageEvent {
    public final String message;
    public MessageEvent(String message) {
        this.message = message;
    }
}
複製代碼

2.註冊觀察者並訂閱事件

選擇要訂閱該事件的訂閱者（subscriber），Activity即在onCreate()加入，調用EventBus的register方法，註冊。數據結構

EventBus.getDefault().register(this);

複製代碼

在不須要接收事件發生時能夠app

EventBus.getDefault().unregister(this);
複製代碼

在訂閱者裏須要用註解關鍵字 @Subscribe來告訴EventBus使用什麼方法處理event。框架

@Subscribe
public void onMessageEvent(MessageEvent event) {
    Toast.makeText(this, event.message, Toast.LENGTH_SHORT).show();
}
複製代碼

注意方法只能被public修飾，在EventBus3.0以後該方法名字就能夠自由的取了，以前要求只能是onEvent().async

3.發送事件

經過EventBus的post方法，發出咱們要傳遞的事件。ide

EventBus.getDefault().post(new MessageEvent("HelloEveryone"));

複製代碼

這樣選擇的Activity就會接收到該事件，而且觸發onMessageEvent方法。函數

EventBus源碼解析

瞭解了對於EventBus的基礎使用，解析來，咱們針對其基礎使用的調用流程，來了解EventBus的實現流程和源碼細節。oop

註冊觀察者

EventBus.getDefault().register(this);
複製代碼

getDefault（）

EventBus.getDefault()是一個單例，實現以下：

public static EventBus getDefault() {  
   if (defaultInstance == null) {  
       synchronized (EventBus.class) {  
           if (defaultInstance == null) {  
               defaultInstance = new EventBus();  
           }  
       }  
   }  
   return defaultInstance;  
} 
複製代碼

保證了App單個進程中只會有一個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);
        }
    }
}
複製代碼

register方法中，首先得到訂閱實例的類，而後調用SubscriberMethodFinder實例的findSubscriberMethods方法來找到該類中訂閱的相關方法，而後對這些方法調用訂閱方法。註冊的過程涉及到兩個問題，一個是如何查找註冊方法？另外一個是如何將這些方法進行存儲，方便後面的調用？

SubscriberMethodFinder是如何從實例中查找到相關的註冊方法的呢？

List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
	//根據類信息叢緩存中查找訂閱方法
    List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
    if (subscriberMethods != null) {
        return subscriberMethods;
    }
	
	//查找註冊方法	
    if (ignoreGeneratedIndex) {
        subscriberMethods = findUsingReflection(subscriberClass);
    } else {
        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;
    }
}
複製代碼

首先從緩存的方法中，經過Class做爲Key進行查找，如何查找內容爲空，則會調用findUsingReflection或者findUsingInfo來從相關類中查找，獲得註冊的方法列表以後，將其添加到緩存之中。

緩存的數據結構以下：

Map<Class<?>, List<SubscriberMethod>> METHOD_CACHE = new ConcurrentHashMap<>();
複製代碼

訂閱方法

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
	 //獲取訂閱方法要監聽的事件類型
    Class<?> eventType = subscriberMethod.eventType;
    
    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);
        }
    }

	//得到該事件類型的訂閱者列表，根據其優先級肯定當前插入者的位置
    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;
        }
    }

	//在該註冊者中加入對應的監聽事件類型
    List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
    if (subscribedEvents == null) {
        subscribedEvents = new ArrayList<>();
        typesBySubscriber.put(subscriber, subscribedEvents);
    }
    subscribedEvents.add(eventType);

	//黏性事件處理
    if (subscriberMethod.sticky) {
        if (eventInheritance) {
            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);
        }
    }
}
複製代碼

subscribe方法的執行流程是先根據事件類型，判斷該註冊者是否已經進行過註冊，若是未註冊將其中的方法進行保存，以事件類型爲鍵保存一份，而後以註冊者實例爲鍵保存一份。

發送事件

對於事件的發送，調用的是post函數

post(Object event)

public void post(Object event) {
	//獲取當前線程的Event隊列，並將其添加到隊列中
    PostingThreadState postingState = currentPostingThreadState.get();
    List<Object> eventQueue = postingState.eventQueue;
    eventQueue.add(event);
    //若是當前PostingThreadState不是在post 中
    if (!postingState.isPosting) {
        postingState.isMainThread = isMainThread();
	    postingState.isPosting = true;
	    if (postingState.canceled) {
	        throw new EventBusException("Internal error. Abort state was not reset");
	    }
	    try {
	    	//遍歷事件隊列，調用postSingleEvent方法
	        while (!eventQueue.isEmpty()) {
	            postSingleEvent(eventQueue.remove(0), postingState);
	        }
	    } finally {
	        postingState.isPosting = false;
	        postingState.isMainThread = false;
	    }  
     }
}
複製代碼

post方法中，首先從當前的PostingThreadState中獲取當前的事件隊列，而後將要post的事件添加到其中，以後判斷當前的線程是否處在post中，若是不在，那麼則會遍歷事件隊列，調用postSingleEvent將其中的事件拋出。

currentPostingThreadState是一個ThreadLocal類型的，裏面存儲了PostingThreadState。

private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {  
       @Override  
       protected PostingThreadState initialValue() {  
           return new PostingThreadState();  
       }  
   }
複製代碼

PostingThreadState包含了一個eventQueue和一些標誌位。類具體結構以下。

final static class PostingThreadState {
    final List<Object> eventQueue = new ArrayList<>();
    boolean isPosting;
    boolean isMainThread;
    Subscription subscription;
    Object event;
    boolean canceled;
}
複製代碼

postSingleEvent

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) {
            logger.log(Level.FINE, "No subscribers registered for event " + eventClass);
        }
        if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
                eventClass != SubscriberExceptionEvent.class) {
            post(new NoSubscriberEvent(this, event));
        }
    }
}
複製代碼

經過lookupAllEventTypes(eventClass)獲得當前eventClass的Class，以及父類和接口的Class類型，然後逐個調用postSingleEventForEventType方法。事件派發的核心方法在postSingleEventForEventType方法中。

postSingleEventForEventType

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;
}
複製代碼

從subscriptionsByEventType中拿到訂閱了eventClass的訂閱者列表，遍歷，調用postToSubscription方法，逐個將事件拋出。

postToSubscription

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 MAIN_ORDERED:
            if (mainThreadPoster != null) {
                mainThreadPoster.enqueue(subscription, event);
            } else {
                invokeSubscriber(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);
    }
}
複製代碼

根據threadMode去判斷應該在哪一個線程去執行該方法,而invokeSubscriber方法內經過反射調用函數。

MainThread

首先去判斷當前若是是UI線程，則直接調用；不然， mainThreadPoster.enqueue(subscription, event) BackgroundThread

若是當前非UI線程，則直接調用；若是是UI線程，則調用backgroundPoster.enqueue方法。

Async

調用asyncPoster.enqueue方法

接下來會針對這幾種廣播方式展開分析

invokeSubscriber

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);
    }
}
複製代碼

經過反射的方式，直接調用訂閱該事件方法。

mainThreadPoster.enqueue

mainThreadPoster = mainThreadSupport != null ? mainThreadSupport.createPoster(this) : null;

複製代碼

mainThreadPoster 經過mainThreadSupport.createPoster建立。

public Poster createPoster(EventBus eventBus) {
    return new HandlerPoster(eventBus, looper, 10);
}
複製代碼

返回HandlerPoster實例。

經過Subscription和Event實例構造出PendingPost，而後將其加入到PendingPostQueue之中，而後調用sendMessage，其handleMessage函數將會被回調。

public 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;
            if (timeInMethod >= maxMillisInsideHandleMessage) {
                if (!sendMessage(obtainMessage())) {
                    throw new EventBusException("Could not send handler message");
                }
                rescheduled = true;
                return;
            }
        }
    } finally {
        handlerActive = rescheduled;
    }
}
複製代碼

當獲得消息以後，開啓循環，從隊列中取PendingPost，調用invokeSubscriber方法執行。

void invokeSubscriber(PendingPost pendingPost) {
    Object event = pendingPost.event;
    Subscription subscription = pendingPost.subscription;
    PendingPost.releasePendingPost(pendingPost);
    if (subscription.active) {
        invokeSubscriber(subscription, event);
    }
}
複製代碼

這裏調用了releasePendingPost

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); } } } 複製代碼

爲了不對象的重複建立，在PendingPost中維護了一個PendingPost列表，方便進行對象的複用。

List<PendingPost> pendingPostPool = new ArrayList<PendingPost>();

複製代碼

對於對象的建立，能夠經過其obtainPendingPost方法來得到。

asyncPoster.enqueue

public void enqueue(Subscription subscription, Object event) {
    PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
    queue.enqueue(pendingPost);
    eventBus.getExecutorService().execute(this);
}
複製代碼

將PendingPost添加到PendingPost隊列中，線程池會從隊列中取數據，而後執行。

@Override
public void run() {
    PendingPost pendingPost = queue.poll();
    if(pendingPost == null) {
        throw new IllegalStateException("No pending post available");
    }
    eventBus.invokeSubscriber(pendingPost);
}
複製代碼

backgroundPoster.enqueue

相比於asyncPoster，backgroundPoster能夠保證添加進來的數據是順序執行的，經過同步鎖和信號量的方式來保證，只有一個線程是在活躍從事件隊列中取事件，而後執行。

public void enqueue(Subscription subscription, Object event) {
    PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
    synchronized (this) {
        queue.enqueue(pendingPost);
        if (!executorRunning) {
            executorRunning = true;
            eventBus.getExecutorService().execute(this);
        }
    }
}
複製代碼

public void run() {
    try {
        try {
            while (true) {
                PendingPost pendingPost = queue.poll(1000);
                if (pendingPost == null) {
                    synchronized (this) {
                        pendingPost = queue.poll();
                        if (pendingPost == null) {
                            executorRunning = false;
                            return;
                        }
                    }
                }
                eventBus.invokeSubscriber(pendingPost);
            }
        } catch (InterruptedException e) {
            
        }
    } finally {
        executorRunning = false;
    }
}
複製代碼

函數掃描

在register方法中對於訂閱方法的查找，調用的方法是SubscriberMethodFinder的findSubscriberMethods方法，對於其中方法的查找有兩種方式，一個是findUsingInfo，一個是findUsingReflection。

private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
	//獲取FindState實例
    FindState findState = prepareFindState();
    findState.initForSubscriber(subscriberClass);
    //從當前類中查找，而後跳到其父類，繼續查找相應方法
    while (findState.clazz != null) {
        findUsingReflectionInSingleClass(findState);
        findState.moveToSuperclass();
    }
    return getMethodsAndRelease(findState);
}
複製代碼

首先，會得到一個FindState實例，其用來保存查找過程當中的一些中間變量和最後結果，首先找當前類中的註冊方法，而後跳到其父類之中，其父類會自動過濾掉Java，Android中的相應類，而後繼續查找。

查找的核心實如今方法findUsingReflectionInSingleClass中。

private void findUsingReflectionInSingleClass(FindState findState) {
    Method[] methods;
    try {
        // 獲取該類中的全部方法，不包括繼承的方法
        methods = findState.clazz.getDeclaredMethods();
    } catch (Throwable th) {
        methods = findState.clazz.getMethods();
        findState.skipSuperClasses = true;
    }
    //遍歷獲取的方法，判斷添加規則爲是否爲public函數，其參數是否只有一個，獲取其註解，而後調用checkAdd，
    //在加入到訂閱方法以前
    for (Method method : methods) {
        int modifiers = method.getModifiers();
        if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
            Class<?>[] parameterTypes = method.getParameterTypes();
            if (parameterTypes.length == 1) {
                Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
                if (subscribeAnnotation != null) {
                    Class<?> eventType = parameterTypes[0];
                    if (findState.checkAdd(method, eventType)) {
                        ThreadMode threadMode = subscribeAnnotation.threadMode();
                        findState.subscriberMethods.add(new SubscriberMethod(method, eventType, threadMode,
                                subscribeAnnotation.priority(), subscribeAnnotation.sticky()));
                    }
                }
            } else if (strictMethodVerification && method.isAnnotationPresent(Subscribe.class)) {
					//多於一個參數
            }
        } else if (strictMethodVerification && method.isAnnotationPresent(Subscribe.class)) {
            //非public，abstract，非靜態的
        }
    }
}
複製代碼

按照以下掃描規則，對類中的函數進行掃描掃描規則：1.函數非靜態，抽象函數 2.函數爲public；3.函數僅單個參數；4.函數擁有@Subscribe的註解;

在符合了以上規則以後，還不可以直接將其加入到函數的隊列之中，還須要對方法進行校驗。

boolean checkAdd(Method method, Class<?> eventType) {
   
    Object existing = anyMethodByEventType.put(eventType, method);
    if (existing == null) {
        return true;
    } else {
        if (existing instanceof Method) {
            if (!checkAddWithMethodSignature((Method) existing, eventType)) {
                
                throw new IllegalStateException();
            }
            anyMethodByEventType.put(eventType, this);
        }
        return checkAddWithMethodSignature(method, eventType);
    }
}

 //函數簽名校驗，來進行
private boolean checkAddWithMethodSignature(Method method, Class<?> eventType) {
    methodKeyBuilder.setLength(0);
    methodKeyBuilder.append(method.getName());
    methodKeyBuilder.append('>').append(eventType.getName());

    String methodKey = methodKeyBuilder.toString();
    Class<?> methodClass = method.getDeclaringClass();
    Class<?> methodClassOld = subscriberClassByMethodKey.put(methodKey, methodClass);
    if (methodClassOld == null || methodClassOld.isAssignableFrom(methodClass)) {
        // Only add if not already found in a sub class
        return true;
    } else {
        // Revert the put, old class is further down the class hierarchy
        subscriberClassByMethodKey.put(methodKey, methodClassOld);
        return false;
    }
}

複製代碼

爲掃描到的函數作校驗，在校驗後，釋放本身持有的資源。第一層校驗在checkAdd函數中，若是當前還沒有有函數監聽過當前事件，就直接跳過第二層檢查。第二層檢查爲完整的函數簽名的檢查，將函數名與監聽事件類名拼接做爲函數簽名，若是當前subscriberClassByMethodKey中不存在相同methodKey時，返回true，檢查結束；若存在相同methodKey時，說明子類重寫了父類的監聽函數，此時應當保留子類的監聽函數而忽略父類。因爲掃描是由子類向父類的順序，故此時應當保留methodClassOld而忽略methodClass。

上述的方式是經過在運行期經過註解處理的方式進行的，效率是比較慢的，在EventBus最新版中引入了在編譯器經過註解處理器，在編譯器生成方法索引的方式進行，以此來提高效率。

粘性事件處理

粘性事件的設計初衷是，在事件的發出早於觀察者的註冊，EventBus將粘性事件存儲起來，在觀察者註冊後，將其發出。經過其內部的一個數據結構：

Map<Class<?>, Object> stickyEvents 
複製代碼

保存每一個Event類型的最近一次post出的event

public void postSticky(Object event) {
    synchronized (stickyEvents) {
        stickyEvents.put(event.getClass(), event);
    }
    // Should be posted after it is putted, in case the subscriber wants to remove immediately
    post(event);
}
複製代碼

將粘性事件保存在stickyEvents，然後post出，此時若是存在已經註冊的觀察者，則狀況同普通事件狀況相同；如尚無註冊的觀察者，在postSingleEvent函數中將時間轉化爲一個NoSubscriberEvent事件發出，可由EventBus消耗並處理。待觀察者註冊時，從stickyEvents中將事件取出，從新分發給註冊的觀察者。

if (subscriberMethod.sticky) {
    if (eventInheritance) {
        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);
    }
}
複製代碼

在對於粘性事件處理這段代碼中，首先判斷是否監聽Event的子類，然後調用checkPostStickyEventToSubscription將黏性事件發出，在checkPostStickyEventToSubscription中，判空後按一半事件的post流程將事件傳遞給觀察者。

private void checkPostStickyEventToSubscription(Subscription newSubscription, Object stickyEvent) {
    if (stickyEvent != null) {	       
        postToSubscription(newSubscription, stickyEvent, isMainThread());
    }
}
複製代碼

小結

輪子的每週一篇，已經到了第四周了，下週是對OkHttp的更細緻的一個剖析，而後是對於OkIO的剖析