EventBus for Android 源碼分析

上文實例講解EventBus for Android介紹了EventBus的基本用法，本文將介紹EventBus的實現原理。EventBus的實現主要圍繞兩個函數register和post,下面分別介紹之。

1 register（Object subscriber)

功能
注冊subscriber中以onEvent開頭的方法
代碼：

private synchronized void register(Object subscriber, boolean sticky, int priority) {
    //構造出方法列表
    List subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriber.getClass());
    for (SubscriberMethod subscriberMethod : subscriberMethods) {
        //注冊每個方法
        subscribe(subscriber, subscriberMethod, sticky, priority);
    }
}

重要數據結構：
SubscriberMethod:存儲方法、線程模型和Event事件處理的參數類型
subscriber函數用兩個Map:subscriptionByEventType和typesBySubscriber來維護事件訂閱者和事件類型的關系，如代碼所示

// Must be called in synchronized block
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod, boolean sticky, int priority) {
    Class eventType = subscriberMethod.eventType;
    CopyOnWriteArrayList subscriptions = subscriptionsByEventType.get(eventType);
    Subscription newSubscription = new Subscription(subscriber, subscriberMethod, priority);
    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);
        }
    }

    // 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> subscribedEvents = typesBySubscriber.get(subscriber);
    if (subscribedEvents == null) {
        subscribedEvents = new ArrayList>();
        typesBySubscriber.put(subscriber, subscribedEvents);
    }
    subscribedEvents.add(eventType);

    ...
}

這個函數中用到了CopyOnWriteArrayList,下次研究一下。

2 post(Object event)

先看源碼：

public void post(Object event) {
    //postingState是一個ThreadLocal型變量， 表示發送事件的進程狀態
    PostingThreadState postingState = currentPostingThreadState.get();
    //將事件加如到隊列中
    List

這裡遇到一個類：PostingThreadState,用來記錄當前線程的狀態和任務隊列，EventBus用postSingleEvent來處理任務隊列中的每個任務，代碼如下：

private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
    Class eventClass = event.getClass();
    boolean subscriptionFound = false;
    if (eventInheritance) {
        List> 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));
        }
    }
}

這個函數的主要功能是獲取Event的類型，如果eventInheritance為True,則會獲取Event的父類，和接口，然後對每一種類型調用一次 postSingleEventForEventType(Object event, PostingThreadState postingState, Class

private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class eventClass) {
    CopyOnWriteArrayList subscriptions;
    synchronized (this) {
        //獲取相應事件類型為eventClass的訂閱者
        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;
}

在獲取到訂閱者，方法和發送事件的線程類型之後，EventBus調用 postToSubscription(Subscription subscription, Object event, boolean isMainThread)來通知訂閱者進行事件處理，該函數代碼如下：

    private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
        switch (subscription.subscriberMethod.threadMode) {
            case PostThread:
                //直接調用Event處理事件
                invokeSubscriber(subscription, event);
                break;
            case MainThread:
                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);
        }
    }

該函數根據不同的線程模型來采用不同的方法來通知訂閱者，下面分別討論之：

1 PostThread模型

直接調用invokeSubscriber(Subscription subscription, Object event)，該方法利用反射原理來調用接受者的事件處理方法

2 MainThread模型

若發送線程為主線程，則直接調用訂閱者事件處理方法。
若發送線程為非主線程，則將事件發送到mainThreadPoster中進行處理。mainThreadPoster的類型為HandlerPoster,繼承自Handler。HandlerPoster內部維護一個隊列（PendingPostQueue）來存取訂閱者和對應的事件響應方法。每次往HandlerPoster中插入數據時， HandlerPoster便發一個消息，告知Handler來從PendingPostQueue中取出數據，然後調用訂閱者的事件處理方法，代碼如下：

void enqueue(Subscription subscription, Object event) {
    PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
    synchronized (this) {
        queue.enqueue(pendingPost);
        if (!handlerActive) {
            handlerActive = true;
            //若looper不在運行，則發消息通知它
            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;
    }
}

3 BackgroundThread模型

若發送線程為非主線程，則直接調用事件響應函數。
若發送線程為主線程，則將消息發送到backgroundPoster中。backgroundPoster的類型為BackgroundPoster,繼承自Runnable。和HandlerPoster類型，該對象也維護一個事件隊列：PendingPosteQueue,每次插入數據時，BackgroundPoster會將自己放入到EventBus的線程池中等待調度，完整代碼如下：

public void enqueue(Subscription subscription, Object event) {
    PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
    synchronized (this) {
        queue.enqueue(pendingPost);
        //若線程池中的任務已經運行，則立即返回，
        if (!executorRunning) {
            executorRunning = true;
            //獲取到EventBus的線程池
            eventBus.getExecutorService().execute(this);
        }
    }
}

@Override
public void run() {
    try {
        try {
            while (true) {
                //等待一秒後取出任務
                PendingPost pendingPost = queue.poll(1000);
                if (pendingPost == null) {
                    synchronized (this) {
                        // Check again, this time in synchronized
                        pendingPost = queue.poll();
                        if (pendingPost == null) {
                            executorRunning = false;
                            return;
                        }
                    }
                }
                eventBus.invokeSubscriber(pendingPost);
            }
        } catch (InterruptedException e) {
            Log.w("Event", Thread.currentThread().getName() + " was interruppted", e);
        }
    } finally {
        executorRunning = false;
    }
}

4 Async模型

該模型直接將事件發送到asyncPoster中進行處理。asyncPoster的類型為AsyncPoster，繼承自Runnable。與BackgroundPoster不同的是，AsyncPoster將數據插入隊列後，直接將自己放入線程池中處理，完成代碼如下：

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

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

Async與BackgroundPoster的不同之處為前者每次加入訂閱者和事件後，便立即將自己放進線程池中，確保每一個任務都處於不同的線程中。而BackgroundPoster會根據線程池中的BackgroundPoster類型的任務是否處於運行狀態來往線程池中加任務，這樣就確保所有處於BackgroundThread模型中的事件處理任務都在同一個後台線程中運行。