分析EventBus源碼擴展Weex事件機制

EventBus 是基于觀察者模式的發(fā)布/訂閱事件總線，它讓組件間的通信變得更加簡單。類似廣播系統(tǒng)，不過 EventBus 所有的訂閱和發(fā)送都是在內(nèi)存層面的，使用起來遠比廣播簡單，也更容易管理。

EventBus-Publish-Subscribe.png

先說明在事件總線中的幾個關(guān)鍵詞：

• 事件發(fā)送者，發(fā)出事件的人
• 訂閱者，處理事件的人
• 訂閱者中處理事件的方法，因為每個訂閱者感興趣的事件有多種，因此會有多個處理事件的方法
• 訂閱，一個訂閱指的是某個訂閱者中的處理某個事件的方法，由訂閱者和事件類型唯一確定。

訂閱事件注冊

當(dāng)希望接受到事件時，需要在 onCreate() 執(zhí)行 register() 方法，在注冊方法中會檢索當(dāng)前類中聲明的接受事件的方法，并將他們注冊到對應(yīng)的映射中。

public void register(Object subscriber) {
    Class<?> subscriberClass = subscriber.getClass();
    List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
    synchronized (this) {
        for (SubscriberMethod subscriberMethod : subscriberMethods) {
            subscribe(subscriber, subscriberMethod);
        }
    }
}

內(nèi)存中存儲的數(shù)據(jù)結(jié)構(gòu)有如下幾個：

// 事件 - List<訂閱(Subscription)> 每個訂閱由訂閱者、事件類型唯一確定
private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;
// 訂閱者 - List<關(guān)注的事件> 每個訂閱者可能關(guān)注多個事件
private final Map<Object, List<Class<?>>> typesBySubscriber;
// 事件對應(yīng)下的粘滯事件
private final Map<Class<?>, Object> stickyEvents;

查找訂閱方法列表

當(dāng)執(zhí)行 register() 方法時，會借助 SubscriberMethodFinder 類從注冊的對象的 Class 中查找。

List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
    //  從緩存中找是否已經(jīng)檢索過了，有緩存就直接返回
    List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
    if (subscriberMethods != null) {
        return subscriberMethods;
    }
    // 是否忽略索引功能，忽略的話會直接使用反射的方法搜索，否則會檢測有沒有相關(guān)的索引可以使用
    if (ignoreGeneratedIndex) {
        subscriberMethods = findUsingReflection(subscriberClass);
    } else {
        // 支持索引的情況，會優(yōu)先從索引中查找，加快查找的速度
        subscriberMethods = findUsingInfo(subscriberClass);
    }
    if (subscriberMethods.isEmpty()) {
        // 沒有找到任何的訂閱方法將會拋出異常，所以至少要用注解訂閱一個方法
    } else {
        // 針對這個 class 查找到訂閱的方法列表，存緩存，下次更快的返回
        METHOD_CACHE.put(subscriberClass, subscriberMethods);
        return subscriberMethods;
    }
}

因為我們不考慮索引的情況，最終查找方法都會走到方法 findUsingReflectionInSingleClass，內(nèi)部的原理相對簡單，遍歷該類的所有方法，找到共有的、只有一個參數(shù)、且?guī)в?@Subscribe 注解的方法，存儲到列表中。

private static final int MODIFIERS_IGNORE = Modifier.ABSTRACT | Modifier.STATIC | BRIDGE | SYNTHETIC;

private void findUsingReflectionInSingleClass(FindState findState) {
    Method[] methods;
    methods = findState.clazz.getDeclaredMethods();    
    for (Method method : methods) {
        int modifiers = method.getModifiers();
        // 共有的方法 & 不是靜態(tài)、抽象、不是編譯生成的方法
        if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
            Class<?>[] parameterTypes = method.getParameterTypes();
            // 參數(shù)長度只能是1
            if (parameterTypes.length == 1) {
                Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
                // 方法上面帶有 @Subscribe 注解
                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()));
                    }
                }
            }
        }
    }
}

這個過程是一個循環(huán)，每次都會向上查找當(dāng)前類的父類，知道到達 java 內(nèi)置的類中，這就意味著，父類中聲明的訂閱方法，在子類實例中也會接收到。查找的結(jié)果最終會生成一個 SubscriberMethod 的列表，這個類中存儲了訂閱方法的全部信息，數(shù)據(jù)結(jié)構(gòu)如下：

public class SubscriberMethod {
    final Method method; // 當(dāng)前的方法，可執(zhí)行
    final ThreadMode threadMode; // 線程類型
    final Class<?> eventType; // 參數(shù)的類型，也就是他訂閱的事件的類型
    final int priority; // 優(yōu)先級
    final boolean sticky; // 是否是粘滯事件
    String methodString; // 方法的字符串
}

訂閱到映射中

// 事件 - List<訂閱(Subscription)> 每個訂閱由訂閱者、事件類型唯一確定
private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;
// 訂閱者 - List<關(guān)注的事件> 
private final Map<Object, List<Class<?>>> typesBySubscriber;

訂閱的過程就是根據(jù)訂閱者 Subscriber 及該訂閱者的某個處理事件的方法 SubscriberMethod 來生成 Subscription 并且存儲到映射當(dāng)中。

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
    // 存儲到 事件 - List<訂閱> 映射中
    Class<?> eventType = subscriberMethod.eventType;
    Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
    CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType); // ... 不存在則創(chuàng)建新的
    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<關(guān)注的事件> 映射中
    List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber); // ... 不存在則創(chuàng)建新的
    subscribedEvents.add(eventType);
    // ...
    // 對 Sticky Event 的處理，后面單獨說
}

取消注冊

由于事件總線的機制基于內(nèi)存實現(xiàn)，所有的訂閱都會存儲在內(nèi)存中，因此必須在合適的時機取消注冊，來釋放占用的內(nèi)存空間。

當(dāng)取消注冊時：

• 借助之前存儲的 訂閱者-List<關(guān)注事件> 的映射快速的獲取到，當(dāng)前訂閱者感興趣的事件列表。
• 然后遍歷事件列表，從 事件-List<訂閱> 的映射中，刪除所有的訂閱。
• 最后將當(dāng)前訂閱者從 訂閱者-List<關(guān)注事件> 刪除，完成取消訂閱的過程。

獲取當(dāng)前訂閱者關(guān)注的全部事件，遍歷取消注冊。

public synchronized void unregister(Object subscriber) {
    List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
    if (subscribedTypes != null) {
        for (Class<?> eventType : subscribedTypes) {
            unsubscribeByEventType(subscriber, eventType);
        }
        typesBySubscriber.remove(subscriber);
    } else {
    }
}

// 從訂閱列表中刪除對應(yīng)的訂閱
private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
    List<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
    if (subscriptions != null) {
        int size = subscriptions.size();
        for (int i = 0; i < size; i++) {
            Subscription subscription = subscriptions.get(i);
            if (subscription.subscriber == subscriber) {
                subscription.active = false;
                subscriptions.remove(i);
                i--;
                size--;
            }
        }
    }
}

發(fā)送事件

當(dāng)需要發(fā)送事件使用 EventBus 的 post() 方法。

借助 ThreadLocal 每個線程單獨維護一個、且僅一個 PostingThreadState 對象，這個對象的數(shù)據(jù)結(jié)構(gòu)如下, 內(nèi)部存儲了當(dāng)前發(fā)送事件狀態(tài)的的一些關(guān)鍵信息。

final static class PostingThreadState {
    final List<Object> eventQueue = new ArrayList<Object>(); // 事件隊列
    boolean isPosting; // 是否正在發(fā)送事件，是的話不需要啟動循環(huán)讀取事件
    boolean isMainThread; // 是否是主線程
    Subscription subscription; // 一個訂閱
    Object event; // 當(dāng)前的事件
    boolean canceled; // 是否被取消
}

獲取本線程的 PostingThreadState 對象，進行初始化，并開始輪詢處理隊列中的事件。

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;
        try {
            // 從隊列中循環(huán)讀取事件處理
            while (!eventQueue.isEmpty()) {
                postSingleEvent(eventQueue.remove(0), postingState);
            }
        } finally {
            postingState.isPosting = false;
            postingState.isMainThread = false;
        }
    }
}

繼續(xù)往深里面看 postSingleEvent() 方法，他每次處理一個從隊列中取出來的事件，這里做了一個區(qū)分，是否支持繼承，這個值默認是 true，支持繼承時，如果對當(dāng)前事件的父類、接口對應(yīng)的事件感興趣，那么他也可以處理該事件。例如當(dāng)前要處理 A 事件，A 繼承自 B，同時實現(xiàn) C 接口，能處理 B,C 事件的訂閱者將也會參與處理此 A 事件。

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) {
        // 沒有找到訂閱的方法，處理分支
    }
}

事件訂閱者排隊處理

接下來會走 postSingleEventForEventType() 方法，這個方法負責(zé)找到對這個事件感興趣的 訂閱 Subscription 列表， Subscription 里面包含了訂閱者、處理對應(yīng)事件的方法等信息。

拿到列表之后便循環(huán)將事件給列表中的訂閱依次處理，在之前注冊時，是有一個優(yōu)先級別的，優(yōu)先級高的將會先獲得處理事件的權(quán)利。

優(yōu)先級別較高的處理者可以停止事件的傳遞，只需要拋出一個異常，被 finally 塊捕捉后，就會中斷輪詢，從而終止事件的傳遞。

private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?>
    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 {
                // 讓 subscription 處理 event
                postToSubscription(subscription, event, postingState.isMainThread);
                aborted = postingState.canceled;
            } finally {
                // 如果優(yōu)先級別較高的處理者異常，則后續(xù)處理者將無法處理該事件
                postingState.event = null;
                postingState.subscription = null;
                postingState.canceled = false;
            }
            // 退出輪詢
            if (aborted) {
                break;
            }
        }
        return true;
    }
    return false;
}

分發(fā)線程處理者執(zhí)行

處理事件的最后一步，是 postToSubscription() 他負責(zé)將事件的處理分發(fā)到不同的線程隊列中，在添加訂閱注解 @Subscribe 時可以指定 threadMode，這極大的方便了我們在事件傳遞后切換不同線程處理事件，例如我們常常要在子線程處理數(shù)據(jù)，而通知主線程更新 UI，使用 EventBus 只需要指定 @Subscribe(threadMode=ThreadMode.Main) 則在處理事件時所有操作在內(nèi)部便被切換到了主線程，真正做到了對線程切換的無感知。

分為了如下幾種類型：

• POSTING 發(fā)送線程，或者說是當(dāng)前線程更貼切一些，在其他類庫中通常叫 Immediate， 也就是不用切換線程。
• MAIN 主線程，不解釋。
• BACKGROUND 后臺線程，如果發(fā)送線程是主線程，則開辟新的線程執(zhí)行，否則將在當(dāng)前線程執(zhí)行。
• ASYNC 異步線程，無論怎樣，總是開啟新的子線程去執(zhí)行。

這里就要看一下幾個處理者 HandlerPoster/BackgroundPoster/AsyncPoster 實現(xiàn)原理大致相同，內(nèi)部維護一個隊列，不停的把里面的事件取出來處理。

• HandlerPoster 是基于 Handler 實現(xiàn)對隊列的輪詢。
• BackgroundPoster 則是用死循環(huán)來做的，誰讓人家有自己的線程呢。
• AsyncPoster 就更富了，根本不輪詢，每次都是一個新的線程。

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;
    }
}

最終調(diào)用的 invokeSubscriber() 很簡單就是利用反射調(diào)一下對應(yīng)的 method

subscription.subscriberMethod.method.invoke(subscription.subscriber, event);

粘滯事件的實現(xiàn)

我把 Sticky Event 翻譯成 粘滯事件 不知道對不對，他的出現(xiàn)主要是因為我們需要處理事件是總是要先注冊再發(fā)送事件，根本原因在于當(dāng)一個事件發(fā)出時，他的生命周期很短，所有對他感興趣的訂閱者處理完了之后他就被拋棄了，后面的訂閱者再感興趣也沒用，因為早就被清理啦。

要解決這個問題也很簡單，就是延長事件的生命周期，即使大家都不理他了，他也能頑強的活著，萬一后面還有人對他感興趣呢。所以實現(xiàn)的原理也就很明了了，找個列表把它全部存起來，除非你手動給刪除，否則就 粘不拉幾 的附著在你的內(nèi)存里，等著他的真命天子出現(xiàn)。

// 事件類型 - 事件實例
private final Map<Class<?>, Object> stickyEvents;
// 發(fā)送粘滯事件時，先存起來給后面的人用，然后按照常規(guī)流發(fā)送出去
public void postSticky(Object event) {
    synchronized (stickyEvents) {
        stickyEvents.put(event.getClass(), event);
    }
    post(event);
}

還要提供一個渠道，讓新加入進來的訂閱者能夠察覺到這里有粘滯事件的存在，如果感興趣也可以處理它。這個時機就是注冊時，當(dāng)一個訂閱者被添加到注冊表中時，此時如果存在粘滯事件，用當(dāng)前訂閱者感興趣的事件為 key 獲取存在的粘滯事件，如果有感興趣的就臨幸一下。于是可以完善一下之前未說完的 register() 方法：

• 首先要求當(dāng)前訂閱者的處理事件的方法要對粘滯事件感興趣，這個在注解上可以聲明。
• 繼承，如果支持繼承，當(dāng)前事件的子類粘滯事件都會被取出來檢查是否可以被處理。

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
    // ... 前面這塊說過了
    // 這個訂閱者的這個訂閱方法是對粘滯事件感興趣的
    if (subscriberMethod.sticky) {
        // 事件是否繼承
        if (eventInheritance) {
            Set<Map.Entry<Class<?>, Object>> entries = stickyEvents.entrySet();
            // 當(dāng)前事件的子類粘滯事件都會被取出來檢查是否可以被處理
            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);
        }
    }
}

接下來的 checkPostStickyEventToSubscription() 就會調(diào)用前面已經(jīng)說過的 postToSubscription() 方法，開始發(fā)送到不同的線程中執(zhí)行，這部分和普通的事件是一樣的啦。

理解事件的繼承

粘滯事件這里也出現(xiàn)了一個關(guān)于事件繼承的檢索，在上一節(jié)也出現(xiàn)了一次，單獨拿出來說一下異同之處。

可以類比函數(shù)入?yún)⒌南拗?，如果一個方法聲明中參數(shù)是父類，那么傳參時可以傳遞子類對象進去，聲明了子類的話，是不能傳遞父類對象的。

舉個例子，設(shè)定下場景，我們現(xiàn)在有事件基類 BaseEvent 和一個事件子類 ImplEvent 是繼承關(guān)系。

第一種場景，發(fā)送普通事件，我發(fā)送了一個 ImplEvent，因為我發(fā)的是個子類事件，也就是說所有聲明關(guān)注 BaseEvent 的訂閱者也都可以將當(dāng)前事件作為入?yún)?，所以向上檢索對 ImplEvent 父類、父接口感興趣的訂閱者去執(zhí)行。

第二個場景，發(fā)送粘滯事件，發(fā)送一個 BaseEvent 的粘滯事件，因為是在注冊時觸發(fā)執(zhí)行，那么說明當(dāng)前訂閱者對 BaseEvent 感興趣，既然他的入?yún)⑹歉割愂录?，那么子類事件也同樣可以作為他的處理事件方法的入?yún)?，于是檢索所有粘滯事件找到所有 BaseEvent 的子類事件都交給當(dāng)前訂閱者處理。

Weex 事件機制

在 Weex 中有一個 BroadcastChannel 的 API 用來實現(xiàn)頁面間的通信，在原生部分使用 WebSocketModule 實現(xiàn)，不過經(jīng)過實驗發(fā)現(xiàn)，注冊和發(fā)送沒有什么大問題，不過在取消注冊這塊做的有漏洞，出現(xiàn)多次頁面銷毀但是無法取消對事件監(jiān)聽的情況（可能是當(dāng)時嘗試的時候版本低一些），主要是因為 module 的生命周期沒能和 weex 頁面實例更好的綁定起來，而且它是基于 W3C 的標(biāo)準(zhǔn)設(shè)計的，也沒有實現(xiàn)類似粘滯事件這種功能的支持。

最后決定根據(jù)事件總線的機制來嘗試實現(xiàn)頁面之間的通信，在 Weex 中有一個 頁面內(nèi) 通信的接口，他是 native 和 weex 通信的通道，可以用一個 key 作為標(biāo)示符，觸發(fā)當(dāng)前 weex 頁面中對 key 事件感興趣的的方法，關(guān)于 weex 相關(guān)的內(nèi)容這里不細說。

((WXSDKInstance)instance).fireGlobalEventCallback(key, params)

實現(xiàn)原理類似 EventBus，不過因為基于 weex 就沒那么復(fù)雜，同樣需要維護一個注冊表，相對于 EventBus 要對訂閱者強引用持有，這里使用了每個 weex 頁面唯一的 instanceId 作為標(biāo)記，存儲這個標(biāo)記而不是存儲真正的 WXSDKInstance 對象，避免內(nèi)存泄漏。

private val mEventInstanceIdMap by lazy { mutableMapOf<String, MutableSet<String>>() }

注冊，當(dāng) weex 那邊發(fā)起注冊時，拿到對應(yīng)的 instanceId 存儲到映射中。

// 注冊接受某事件
// event.registerEvent('myEvent')
// globalEvent.addEventListener('myEvent', (params) => {});
fun registerEvent(key: String?, instantId: String?) {
    // do check...
    val nonNullKey = key ?: return
    val registerInstantIds = mEventInstanceIdMap[nonNullKey] ?: mutableSetOf()
    registerInstantIds.add(instantId)
    mEventInstanceIdMap[nonNullKey] = registerInstantIds
}

發(fā)送事件時，根據(jù)事件的 key 拿到對他關(guān)注的訂閱者的 instanceId 列表，循環(huán)從 weex sdk 中取出真正的 WXSDKInstance 對象，再利用頁面內(nèi)通信的 API 將事件發(fā)送給指定頁面，達到頁面間通信的目的。

// 發(fā)送事件
// event.post('myEvent',{isOk:true});
fun postEvent(key: String, params: Map<String, Any>) {
    // do check...
    val registerInstantIds = mEventInstanceIdMap[key] ?: listOf<String>()
    val allInstants = renderManager.allInstances
    for (instance in allInstants) {
        // 遍歷找到訂閱的 instanceId 進而拿到 weex 實例發(fā)送頁面內(nèi)事件
        if (instance != null
                && !instance.instanceId.isNullOrEmpty()
                && registerInstantIds.contains(instance.instanceId)) {
            instance.fireGlobalEventCallback(key, params)
        }
    }
}

當(dāng)頁面銷毀時，同時自動取消注冊，釋放內(nèi)存和避免不必要的事件觸發(fā)

override fun onWxInstRelease(weexPage: WeexPage?, instance: WXSDKInstance?) {
    val nonNullId = instance?.instanceId ?: return
    for (mutableEntry in mEventInstanceIdMap) {
        if (mutableEntry.value.isNotEmpty()) {
            mutableEntry.value.remove(nonNullId)
        }
    }
}

最后，目前只是一個簡單的實現(xiàn)，能夠基本實現(xiàn)頁面間通信的需求，不過還需要更多地調(diào)研和其他端同學(xué)的配合，相信會越來越完善。

目前維護的幾個項目，求 ????

• SocialSdk 登錄分享功能原生接入
• LightAdapter 輕量級適配器
• ImageEditor 圖片處理，裁剪旋轉(zhuǎn)，貼紙涂鴉，濾鏡等
• WeexCube Weex 容器方案
• Kotlin 學(xué)習(xí)系列總結(jié),共計 22 篇