EventBus简析

概述

EventBus 是适用于Android与Java的一款发布/订阅式事件总线。官方给出的演示图如下：

EventBus-Publish-Subscribe.png

EventBus有如下特点（翻译自官方文档，语言水平不足，请见谅）：

• 简化各组件之间的通信：
  • 解藕事件发送者和接收者
  • 在Activities、Fragments以及后台线程中表现良好
  • 避免了复杂和易于出错的依赖及生命周期问题
• 简化代码编写
• 更快
• 更小（~50k jar）
• 通过100,000,000+已安装应用的实践证明
• 拥有诸如线程交付、订阅者优先级等高级功能

使用示例

添加依赖库

在app的build.gradle中添加依赖库：

compile 'org.greenrobot:eventbus:3.1.1'

定义事件类

定义需要传输的事件，通常包含get()方法，用以获取传输的对象。

public class MessageEvent {
    MessageBean mMessage;

    public MessageEvent(MessageBean message) {
        this.mMessage = message;
    }

    public MessageBean getMessage() {
        return mMessage;
    }
}

public class MessageBean {
    public String title;
    public String message;
}

订阅

在需要接收事件的组件中，需要注册订阅者，这里我们在Activity的onCreate()中注册：

@Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        ...
        EventBus.getDefault().register(this);
    }

通常有注册的地方，都不要忘了解注册，我们在onCreate()中注册，相应的就应该在onDestroy()中解注册：

@Override
protected void onDestroy() {
    super.onDestroy();
    EventBus.getDefault().unregister(this);
}

注册了之后，需要添加对事件的处理，EventBus 3.0之前是通过约定的方法名来实现的，订阅者需要实现如下方法：

public void onEvent(MessageBean message) {
    ...
}

3.0之后，EventBus改用注解的方式：

@Subscribe(threadMode = ThreadMode.MAIN)
public void onEvent(MessageBean message) {
    mTextView.setText(message.title + "\n" + message.message);
}

发布事件

发布事件十分简单，构建发布对象，调用EventBus的post()方法即可：

MessageBean message = new MessageBean();
message.title = "MessageBean";
message.message = "HelloWorld";                
EventBus.getDefault().post(message);

如此，便完成了一个简单的事件发布、订阅流程，更多详细操作请参阅官方文档。

接下来我们便对EventBus的代码进行解析。

解析

首先看下注册订阅者做了哪些操作。

订阅

EventBus

先来看下EventBus维护的几个映射表：

• Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType：该表维护了一个事件类型到Subscription对象的映射表，事件类型即订阅方法（@Subscribe标识的方法）中的参数类型，Subscription类封装了一个订阅者与一个订阅方法。
• Map<Object, List<Class<?>>> typesBySubscriber：该表维护了一个订阅者到事件类型的映射表。
• Map<Class<?>, Object> stickyEvents：粘性事件类型到事件的映射表（调用postSticky()时用到）。

订阅/发布的时候，我们发现都是通过EventBus.getDefault()方法获取到EventBus实例，再进行相应操作的，可见EventBus使用了单例模式，并且是线程安全的双重锁模式：

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

获取到EventBus实例后，组件需要调用register()方法将自己注册为订阅者，我们来看下register()做了什么：

private void register(Object subscriber) {
    // 获取订阅者Class对象
    Class<?> subscriberClass = subscriber.getClass();
    // 获取订阅者的订阅方法
    List<SubscriberMethod subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
    synchronized (this) {
        for (SubscriberMethod subscriberMethod : subscriberMethods) {
            // 对订阅者中的每个订阅方法进行订阅
            subscribe(subscriber, subscriberMethod):
        }
    }
}

register()方法主要是获取订阅者的订阅方法，然后调用subscirbe()方法进行订阅操作。

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
    Class<?> eventType = subscriberMethod.eventType;
    // Step 1：构建Subscription对象并添加到subscriptionsByEventType表中
    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;
        }
    }

    // Step 2：将事件类型添加到typesBySubscriber表中
    List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
    if (subscribedEvents == null) {
        subscribedEvents = new ArrayList<>();
        typesBySubscriber.put(subscriber, subscribedEvents);
    }
    subscribedEvents.add(eventType);
    // Step 3: 如果是粘性事件(@Subscribe(sticky = true))
    // 且stickyEvents表中存在该事件，则发布粘性事件
    if (subscriberMethod.sticky) {
        ...
    }
}

看下来发现， 其实注册订阅者就是把订阅者的Class对象以及查找到的订阅方法添加到subscriptionsByEventType表以及typesBySubscriber表中。

接下来看下是怎么找到订阅方法的。

SubscriberMethodFinder

SubscriberMethodFinder，顾名思义，订阅方法查找器，通过该类的一些方法调用可以找到订阅者的订阅方法，我们主要看下findSubscriberMethods()方法：

List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
    // 先查看缓存中是否有订阅方法，如果有的话直接返回，不再查找
    List<SubscriberMethod> subscriberMEthods = METHOD_CACHE.get(subscriberClass);
    if (subscriberMethods != null) {
        return subscriberMethods;
    }
    // ignoreGeneratedIndex默认为false
    if (ignoreGeneratedIndex) {
        subscriberMethods = findUsingRefelction(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;
    }
}

·SubscriberMethodFinder·维护了一个METHOD_CACHE映射表，用来存放订阅者到订阅方法的映射。查找订阅方法时，首先会看下缓存中是否有该方法，如果有的话，就不再查找，直接返回缓存的方法列表。
这里需要注意的是，注册了订阅者的类必须实现订阅方法，否则运行时会抛出异常。
查找方法时根据ignoreGeneratedIndex的值决定是调用findUsingReflection()方法还是使用findUsingInfo()方法。ignoreGeneratedIndex默认为false，初始化EventBus时可通过Builder自定义。findUsingReflection()与findUsingInfo()最终调用的都是findUsingReflectionInSingleClass()方法，通过反射来查找订阅方法：

private void  findUsingReflectionInSingleClass(FindState findState) {
    Method[] methods;
    try {
        // 比getMethods()更快，尤其当订阅者是Activitity之类的fat classes(?)
        methods = findState.clazz.getDeclaredMethods();
    } catch (Throwable th) {
        methods = findState.clazz.getMethods();
        findState.skipSuperClasses = true;
    }
    ...
}

首先通过订阅者的类对象获取订阅者的方法列表，接下去就对这些方法进行过滤：

private void  findUsingReflectionInSingleClass(FindState findState) {
    ...
    for (Method method : methods) {
        // 获取方法的修饰符public/private/abstract/...
        int modifiers = method.getModifiers();
        // 只有以public修饰，且非abstract、static或编译器合成的方法
        if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
            Class<?>[] parameterTypes = method.getParameterTypes();
            if (parameterTypes.length == 1) {
                // 获取@Subscribe注解的参数
                Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
                if (subscribeAnnotation != null) {
                    Class<?> eventType = parameterTypes[0];
                    if (findState.checkAdd(method, eventType)) {
                        // 满足这些条件之后，将该方法作为@Subscriber方法添加到列表里
                        ThreadMode threadMode = subscribeAnnotation.threadMode();
                        findState.subscriberMethods.add(new SubscriberMethod(method, eventType, threadMode,
                                    subscribeAnnotation.priority(), subscribeAnnotation.sticky()));
                }
            } 
            // @Subscribe 方法有且仅有1个参数
            else if (strictMethodVerification & method.isAnnotationPresent(Subscribe.class)) {
                String methodName = method.getDeclaringClass().getName() + "." + method.getName();
                throw new EventBusException("@Subscribe method " + methodName +
                            "must have exactly 1 parameter but has " + parameterTypes.length);
            }
        } 
        // @Subscribe 方法不合法，必须是public、非static、非abstract
        else if (strictMethodVerification && method.isAnnotationPresent(Subscribe.class)) {
            String methodName = method.getDeclaringClass().getName() + "." + method.getName();
            throw new EventBusException(methodName +
                        " is a illegal @Subscribe method: must be public, non-static, and non-abstract");
        }
    }
}

由上可以看到我们注册订阅时必须满足以下条件：

1. 订阅者必须定义订阅方法
2. 订阅方法必须是public、非static、非abstract的
3. 订阅方法必须有且仅有一个参数，该参数的类型即为事件类型
4. 订阅方法由@Subscribe注解标识

事件分发

接下来我们来了解下事件是如何分发的

post()

先来看下post()方法

public void post(Object event) {
    PostingThreadState postingState = currentPostingThreadState.get();
    // 事件队列
    List<Object> eventQueue = postingState.eventQueue;
    eventQueue.add(event);

    if (!postingState.isPosting) {
        // 是否主线程
        postingState.isMainThread = isMainThread();
        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;
        }
    }
}

可以看到，EventBus首先会将事件加入到一个eventQueue列表中，然后循环分发事件队列。

private void postSingleEvent(Object event, PostingThreadState postingState) {
    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) {
        ...
    }
}

首先判断是否支持事件继承（默认不支持），如果支持的话，会层层往上获取父类或接口，并对每个父类进行post操作，具体的操作位于postSingleEventForEventType()方法中：

private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
    // 获取该事件类型对应的Subscription对象
    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.subscripiton = null;
                postingState.canceled = false;
            }
            // 如果事件取消了，则中断发送
            if (aborted) {
                break;
            }
        }
        // 找到了事件的订阅者，返回true
        return true;
    }
    // 没有事件相关订阅者，返回false
    return false;
}

postSingleEventForEventType()方法主要是从subscriptionsByEventType映射表中取出事件对应的订阅列表，一一发送事件，发送事件的具体实现在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 {
                // temporary: technically not correct as poster not decoupled from subscriber
                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);
    }
}

这里根据订阅方法@Subscribe注解中threadMode的参数来决定发送的方式，threadMode有以下几种类型：

• POSTING：默认值，订阅者与发送者在同一进程，直接调用订阅方法。
• MAIN：订阅者将在主线程调用，如果发送者也在主线程，则直接调用订阅方法；否则将事件加入队列中等待分发。
• MAIN_ORDERED：订阅者在主线程调用，与MAIN模式的区别在于，事件总是被加入队列中等待分发。
• BACKGROUND：订阅者将在后台线程中调用。如果发送线程不是主线程，则直接分发事件，否则加到队列中等待。
• ASYNC：订阅者将在一个单独的线程中被调用，并且该线程独立于主线程和分发线程。如果订阅者处理耗时长的事件时，建议使用这种模式，例如网络访问。为了防止触发大量异步线程，EventBus采用线程池的方式来管理这些订阅线程。

根据以上的代码可以轻松地理解这几种模式的特点。

接下来我们分别看下几种分发方式。

直接调用

void invokeSubscriber(Subscription subscription, Object event) {
    try {
        subscription.subscriberMethod.method.invoke(subscripiton.subscriber, event);
    } catch (InvocationTargetException e) {
        handleSubscriberException(subscription, event, e.getCause());
    } catch (IllegalAccessException e) {
        throw new IllegalStateException("Unexpected exception", e);
    }
}

直接调用很简单，就是根据之前记录的类名、方法名，通过反射方法直接调用。

主线程分发器

主线程分发器 -- mainThreadPoster，主要用于ThreadMode.MAIN与ThreadMode.MAIN_ORDERED模式，它的类型是HandlerPoster，我们来看下它的实现：

public class HandlerPoster extends Handler implements Poster {
    // 自定义的阻塞队列
    private final PendingPostQueue queue;
    private final int maxMillisInsideHandleMessage;
    private final EventBus eventBus;
    private boolean handlerActive;
    
    protected HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
        super(looper);
        this.eventBus = eventBus;
        this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
        queue = new PendingPostQueue();
    }
  
    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;
                // 当轮循时间等于或大于maxMillisInsideHandleMessage时
                // 重新发送消息
                if (timeInMethod >= maxMillisInsideHandleMessage) {
                    if (!sendMessage(obtainMessage())) {
                        throw new EventBusException("Could not send handler message");
                    }
                    rescheduled = true;
                    return;
                }
            }
        } finally {
            handlerActive = rescheduled;
        }
    }
}

可以看出来，mainThreadPoster 就是使用Handler机制，利用主线程的looper来分发事件。

后台分发器

backgroundPoster用于ThreadMode.BACKGROUND模式，BackgroundPoster类型。与HandlerPoster相似，它也维护了一个PendingPostQueue队列，我们主要看下它的enqueue()与run()方法。

final class BackgroundPoster implements Runnable, Poster {
    ...
    public void enqueue(Subscription subscription, Object event) {
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        synchronized (this) {
            queue.enqueue(pendingPost);
            // 如果线程已在执行，不再添加
            if (!executorRunning) {
                executorRunning = true;
                // 调用EventBus的线程池来处理该Runnable
                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) {
                eventBus.getLogger().log(Level.WARNING, Thread.currentThread().getName() + " was interruppted", e);
            }
        } finally {
            executorRunning = false;
        }
    }
}

由BackgroundPoster的实现可以看出，它是在后台线程调用订阅方法的，并由EventBus的线程池来管理，默认使用的是CachedThreadPool：

private final static ExecutorService DEFAULT_EXECUTOR_SERVICE = 
          Executors.newCachedThreadPool();

异步分发器

asyncPoster用于处理ThreadMode.ASYNC模式的订阅方法，它是AsyncPoster对象，这里也直接看enqueu()与run()方法。

class AsyncPoster implements Runnable, Poster {
    ...
    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);
    }
}

可以看到，AsyncPoster虽然也实现了Runnable，但它与BackgroundPoster最明显的区别在于，调用enqueue()时，backgroundPoster将事件添加到队列中后，会判断当前线程是否已在运行，如果是的话，将不再加入线程池。而AsyncPoster则是每一次调用enqueue()时，都会往线程池添加一个对象，来将当前事件分发到对应的订阅方法中。

解注册

最后再说一下解注册。
我们前面说到，注册订阅者就是把订阅者的Class对象以及查找到的订阅方法添加到subscriptionsByEventType表以及typesBySubscriber表中。那么，很显然，解注册就是把订阅者相关信息从表中删除。

public synchronized void unregister(Object subscriber) {
    List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
    if (subscribedTypes != null) {
        for (Class<?> eventType : subscribedTypes) {
            // 从subscriptionsByEventType中移除相关信息
            unsubscribeByEventType(subscriber, eventType);
        }
        typesBySubscriber.remove(subscriber);
    } else {
        logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
    }
}