EventBus 源码解析

本文目的

---

分析EventBus是如何工作的（基于3.0版本）。

放两张EventBus内部的数据的存储图：

eventbus_eventmethods.png

这里是用一个HashMap保存每一个事件类型对应的所有监听方法。

typesBySubscriber.png

这里是用一个HashMap保存每一个注册了的类中，所有的监听事件的类型。

先看看EventBus.java中，它有哪些属性值得关注：

public class EventBus {

    static volatile EventBus defaultInstance;
    // 保存每个事件类型接收通知的方法
    private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;
    // 保存每个类中，所有的事件类型
    private final Map<Object, List<Class<?>>> typesBySubscriber;
    private final Map<Class<?>, Object> stickyEvents;

    private final HandlerPoster mainThreadPoster;
    private final BackgroundPoster backgroundPoster;
    private final AsyncPoster asyncPoster;
    private final SubscriberMethodFinder subscriberMethodFinder;
    private final ExecutorService executorService;

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

Subscription就是一个类+它内部一个事件监听方法的封装：

final class Subscription {
    final Object subscriber;
    final SubscriberMethod subscriberMethod;
}

SubscribeMethod代表一个事件监听方法

public class SubscriberMethod {
    final Method method;
    final ThreadMode threadMode;
    final Class<?> eventType;
    final int priority;
    final boolean sticky;
    /** Used for efficient comparison */
    String methodString;
}

注册过程

---

我们一般都是在Activity的onStart方法中，执行EventBus.getDefault().register(this):

    public void register(Object subscriber) {
        Class<?> subscriberClass = subscriber.getClass();
        // 这里通过2种方式去获取一个类里面所有加了@Subscribe注解的事件监听方法：
        // 1、是通过annotationProcessor，在编译时获取注解，然后生成Java代码，在运行时叫用此生成的Java代码，完成所有类中监听方法的收集工作。
        // 2、使用反射
        List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
        synchronized (this) {
            for (SubscriberMethod subscriberMethod : subscriberMethods) {
                subscribe(subscriber, subscriberMethod);
            }
        }
    }

这里根据注册的类，找到这个类里所有的EventBus事件方法List<SubscriberMethod>，然后对每一个方法中的事件进行订阅：

    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) {
                // 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);
            }
        }
    }

看看粘性事件的发布:

    private void checkPostStickyEventToSubscription(Subscription newSubscription, Object stickyEvent) {
        if (stickyEvent != null) {
            // If the subscriber is trying to abort the event, it will fail (event is not tracked in posting state)
            // --> Strange corner case, which we don't take care of here.
            postToSubscription(newSubscription, stickyEvent, Looper.getMainLooper() == Looper.myLooper());
        }
    }

看名字，应该就是将事件发布给订阅者了，好了，消息的详细发布过程后面再看，先来看看我们发送普通事件的过程是怎样的。

事件发布

---

一般情况，我们发布事件是调用EventBus.getDefault().post()：

    public void post(Object event) {
        // 又看到了ThreadLocal的身影了，此处currentPostingThreadState.get就是在每个线程内部都保存一份自己的PostingThreadState
        PostingThreadState postingState = currentPostingThreadState.get();
        List<Object> eventQueue = postingState.eventQueue;
        eventQueue.add(event);
       
        // 如果当前线程的事件没有正在发布，则开始发布
        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 {
                // 循环发布此线程中事件队列的事件
                while (!eventQueue.isEmpty()) {
                    postSingleEvent(eventQueue.remove(0), postingState);
                }
            } finally {
                postingState.isPosting = false;
                postingState.isMainThread = false;
            }
        }
    }

其中PostingThreadState是EventBus.java的内部类，它就是维护本线程内的发布状态，主要是事件队列。

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

不过等等，既然是每个线程都有一个PostingThreadState，那应该会有set的地方，就像Looper中一样。找啊找，貌似没在EventBus中找到，难道这个get()方法自带初始化功能？于是点进ThreadLocal的get()方法看看：

    public T get() {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null) {
            ThreadLocalMap.Entry e = map.getEntry(this);
            if (e != null) {
                @SuppressWarnings("unchecked")
                T result = (T)e.value;
                return result;
            }
        }
        return setInitialValue();
    }

emmmm，setInitialValue()，有点意思了，继续看

    private T setInitialValue() {
        T value = initialValue();
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null)
            map.set(this, value);
        else
            createMap(t, value);
        return value;
    }

看到initialValue()方法有没有很眼熟啊，对，就是本文开头中，currentPostingThreadState声明的地方：

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

原来是这样，每次取出当前线程的ThreadPostingState的时候，如果为null，则会先创建一个再返回。好了，继续。

上面post()方法中调用了postSingleEvent()，这个方法可以跳过，没什么好说的，然后又调用了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) {
                // 从这里可以看出，这个ThreadPostingState还真是实时的发送状态啊，每发布一个，就改一次内部的事件类型、订阅者等
                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;
    }

可以看到它循环取出该事件的接受者，一一调用了同发送粘性事件一样的方法，开始发布事件。

好了，接下来，我们该看看之前的具体的事件发布细节了：

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

这段代码很容易理解

• 如果threadMode为POSTING，那么就是接收通知和发送通知要处于同一线程，直接调用接收方法就可以了
• 如果threadMode为MAIN，就先看当前是否在主线程，在就直接调用接收方法；不在，就将此事件加入到主线程发送者的队列中
• 如果threadMode为BACKGROUND，就先判断当前是否主线程，如果是，就加入后台发送者的消息队列中等待发送；如果不是，直接调用
• 如果threadMode为ASYNC，就是说接收线程要和发送线程处在不同的线程中。那么就将事件异步发送者的队列中等待发送。

先看看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);
        }
    }

就只是调用反射而已嘛。
再看看那几个Poster：mainThreadPoster、backgroundPoster和asyncPoster。这几个Poster都是涉及到要跨线程的了。
一提到跨线程，那就很容易想到Handler。是的，mainThreadPoster，它还就是一个Handler：

final class HandlerPoster extends Handler {

    private final PendingPostQueue queue;
    private final int maxMillisInsideHandleMessage;

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

这个Handler很简单，enqueue加入一个封装了事件和订阅者的PendingPost，然后sendMessage()发送一条空消息，告诉handleMessage()方法，嘿嘿嘿，开工了开工了！于是mainThreadPoster开始循环取出自己的事件队列中的PendingPost，并调用eventBus.invokeSubscriber(pendingPost)方法，继续反射调用起事件接收方法。
其中PendingPostQueue就是一个自定义的队列而已。

final class PendingPostQueue {
    private PendingPost head;
    private PendingPost tail;

    // 定义poll()等方法
}

PendingPost也就是对一个事件对封装，包含了接收者

final class PendingPost {
    private final static List<PendingPost> pendingPostPool = new ArrayList<PendingPost>();

    Object event;
    Subscription subscription;
    PendingPost next;
}

这样，主线程的事件分发就完成了。

那么还有BackgroundPoster、AsyncPoster，他们都是实现了Runnable接口，以AsyncPoster为例：

class AsyncPoster implements Runnable {

    private final PendingPostQueue queue;
    private final EventBus eventBus;

    AsyncPoster(EventBus eventBus) {
        this.eventBus = eventBus;
        queue = new PendingPostQueue();
    }

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

}

它们使用线程池来处理加入进来的事件，BackgroundPoster类似。可以看到，它们其实都是借鉴了Android的消息队列机制，在本线程去拉取出一个消息队列中的消息执行，达到跨线程的目的。

这就是EventBus 3.0 一次注册、订阅、发布事件的过程。

使用到的设计模式

---

• 观察者模式
  我们定义的事件作为Observable，那些注册了的类，作为Observer
• 建造者模式
  EventBusBuilder作为Builder角色，而EventBus自己又作为了Director角色
• 单例模式
  EventBus实例的创建使用了经典的DCL+volatile方式实现

总结

---

EventBus的源码还是比较简单的。它的大致流程如下：

• 使用一个HashMap保存每个事件及其注册了的类中的监听方法。
• 使用一个HashMap保存每个注册的类中所有的事件。
• 如果使用APT技术，则在编译时，就获取到了所有加在监听方法上的@Subscribe注解，完成监听方法的收集。
• 如果使用反射技术，则在注册阶段去收集一个类中的监听方法，效率会相对低一些。
• 注册时，就是将事件和监听方法加入到上述两个Map中的过程，并且在加入的时候，如遇粘性事件，会立即发布
• EventBus中一个将线程分为四类：主线程、发布线程、后台线程、异步线程。每个线程都有自己的一个ThreadPostingState，里面保存当前线程的发布状态，包括需要发布的事件队列，正在发送的事件-接受者，等等。
• 发布事件时，会遍历当前线程的ThreadPostingState，取出其事件队列中的事件，循环发送每一个事件。
• 发布一个事件时，会先找到这个事件的所有监听者，然后根据每个监听者的监听模式，采用不同的发布策略。
• 对于不需要切换线程的情形，直接使用反射调用该监听方法，完成事件发布。
• 对于要切换到主线程监听的情形，则使用Handler完成线程切换。
• 对于要切换到后台线程和异步线程的情形，则使用线程池，完成线程切换。