EventBus源码分析

EventBus-Publish-Subscribe.png

简介

Android或者Java平台下的一款高效的事件总线框架。

版本 ： org.greenrobot:eventbus:3.1.1

特点：

• 简洁的组件通信机制
  • 发布者、订阅者完全分离
  • 线程安全
  • 内存安全
• 调用者代码简洁
• 快捷迅速
• 包小
• 经过了100,000,000+App安装量验证
• 支持优先级

源码分析

我们以一个最简单的示例作为分析的突破口（以下代码均以Kotlin编写）。

第1步：编写一个事件类

class MessageEvent(val data: String)

第2步：订阅者

class MainActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)
        //1、注册
        EventBus.getDefault().register(this)
    }

    //2、定义订阅触发方法
    @Subscribe(threadMode = ThreadMode.MAIN)
    fun onMessageEvent(event: MessageEvent) {
        Log.i("test", "MessageEvent事件被触发")
    }

    override fun onDestroy() {
        super.onDestroy()
        //3、注销
        EventBus.getDefault().unregister(this)
    }
}

第3步：发布者

 EventBus.getDefault().post(MessageEvent("123456"))

显然第1步没什么好分析的，怎么看它都是一个普普通通的类。

EventBus.getDefault().register(this) 源码

而第2步中大致又可以分为三部分：注册、定义触发方法、注销。我们看看注册的源码都执行了哪些逻辑：

// EventBus.getDefault().register(this)方法源码
public void register(Object subscriber) {
    //1
    Class<?> subscriberClass = subscriber.getClass();
    //2
    List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
    synchronized (this) {
        //3
        for (SubscriberMethod subscriberMethod : subscriberMethods) {
            subscribe(subscriber, subscriberMethod);
        }
    }
}

register()方法内部实现大致可看为3行代码：

• 1、获取传入subscriber对象的字节码对象
• 2、通过字节码对象获取到一个List<SubscriberMethod>集合
• 3、遍历这个集合，调用subscribe()方法 ，把当前subscriber和item subscriberMethod以形参形式传入

字节码对象不必多讲，我们看看subscriberMethodFinder.findSubscriberMethods()这个方法返回的集合究竟是什么：

List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
    缓存机制...
    
    if (ignoreGeneratedIndex) {
        subscriberMethods = findUsingReflection(subscriberClass);
    } else {
        subscriberMethods = findUsingInfo(subscriberClass);
    }

    缓存机制...
    return subscriberMethods;
}

便于阅读，这里省掉部分不影响核心功能的的代码，重点放到这个if判断上，通过源码的查看ignoreGeneratedIndex我们并没有进行过修改操作，这里走的是默认值false,再看看findUsingInfo()方法做了什么：

private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
    //1、
    FindState findState = prepareFindState();
    findState.initForSubscriber(subscriberClass);
    //2、
    while (findState.clazz != null) {
        findState.subscriberInfo = getSubscriberInfo(findState);
        if (findState.subscriberInfo != null) {
           缓存机制...
        } else {
            findUsingReflectionInSingleClass(findState);
        }
        //3、
        findState.moveToSuperclass();
    }
    //4、
    return getMethodsAndRelease(findState);
}

这里又出现了一个新类FindState：

static class FindState {
    final List<SubscriberMethod> subscriberMethods = new ArrayList<>();
    final Map<Class, Object> anyMethodByEventType = new HashMap<>();
    final Map<String, Class> subscriberClassByMethodKey = new HashMap<>();
    final StringBuilder methodKeyBuilder = new StringBuilder(128);

    Class<?> subscriberClass;
    Class<?> clazz;
    boolean skipSuperClasses;
    SubscriberInfo subscriberInfo;

    void initForSubscriber(Class<?> subscriberClass) {
        this.subscriberClass = clazz = subscriberClass;
        skipSuperClasses = false;
        subscriberInfo = null;
    }
    
    ...
｝

可以看出来FindState是一个容器管理类，貌似存放了一些跟这个订阅者Class字节码对象相关的数据，顺便看下initForSubscriber()方法将我们的订阅者Class对象赋值给了该容器的subscriberClass和class.

再看这里的while循环肯定为true了，if语句我们看else部分，其他部分属于缓存机制，findUsingReflectionInSingleClass()方法如下：

private void findUsingReflectionInSingleClass(FindState findState) {
    //1、
    Method[] methods;
    try {
        // This is faster than getMethods, especially when subscribers are fat classes like Activities
        methods = findState.clazz.getDeclaredMethods();
    } catch (Throwable th) {
        // Workaround for java.lang.NoClassDefFoundError, see https://github.com/greenrobot/EventBus/issues/149
        methods = findState.clazz.getMethods();
        findState.skipSuperClasses = true;
    }
    
    //2、
    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();
                        //3、
                        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)) {
            ...
        }
    }
}

首先通过反射获取到了当前订阅者Class中的所有Method对象，接下来遍历这些Method对象，最终把我们编写订阅触发方法的Method找了出来（@Subscribe注解标识的方法），一个订阅者可以有多个订阅触发方法，都将它们存储到findState.subscriberMethods集合里。

让我们findUsingInfo()，至于findState.moveToSuperclass()我们就不看了，这里就是获取当前订阅者Class的superClass通过while再去查询它里面的订阅事件，重走刚刚所分析的逻辑，最后看getMethodsAndRelease(findState)返回了什么东西：

private List<SubscriberMethod> getMethodsAndRelease(FindState findState) {
    List<SubscriberMethod> subscriberMethods = new ArrayList<>(findState.subscriberMethods);
    findState.recycle();
    缓存机制...
    return subscriberMethods;
}

原来就是将findState.subscriberMethods返回了回去，通过一层层的返回最终subscriberMethods返回到了register()方法中，那么register()方法中就剩最后遍历调用的subscribe()方法，看看它都做了什么：

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
    Class<?> eventType = subscriberMethod.eventType;
    //1、
    Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
    CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
    if (subscriptions == null) {
        subscriptions = new CopyOnWriteArrayList<>();
        //2、
        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) {
            //2、
            subscriptions.add(i, newSubscription);
            break;
        }
    }

    List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
    if (subscribedEvents == null) {
        subscribedEvents = new ArrayList<>();
        typesBySubscriber.put(subscriber, subscribedEvents);
    }
    subscribedEvents.add(eventType);

    粘性事件处理...
}

这里代码看似很多，大家只需记住1个对象、2个容器即可：

• 对象：Subscription
  • 存储了注册时传入的订阅者subscriber 和 某个订阅方法（@Subscribe注解标注的方法）
• 容器1：Map<Class,List< Subscription >> subscriptionsByEventType
  • 存储了某个事件类型（eg:MessageEvent） -> 对应的所有的Subscription（小对象）
• 容器2：Map<Object, List< Class >> typesBySubscriber
  • 存储了某个订阅者 -> 对应订阅的所有事件类型（eg:MessageEvent...）

group.png

到此整个register()的源码已经走了一遍，一句话概括：

register()方法，通过反射将订阅者中@Subscribe注解标注的所有方法、订阅对象、订阅事件类型分别存储到了两个容器中。

EventBus.getDefault().unregister(this) 源码

想想，register()方法是将订阅者拆解出来存储到不同容器中，那么unregister()用来做什么？
那肯定是将当前订阅者相关的存储数据从容器中remove喽：

public synchronized void unregister(Object subscriber) {
    //1、
    List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
    if (subscribedTypes != null) {
        //2、
        for (Class<?> eventType : subscribedTypes) {
            unsubscribeByEventType(subscriber, eventType);
        }
        typesBySubscriber.remove(subscriber);
    } else {
        logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
    }
}

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

可以看到先通过容器typesBySubscriber获取到当前订阅者订阅的所有事件类型，并从该容器中清除，再通过unsubscribeByEventType()方法清除subscriptionsByEventType容器中跟这个订阅者相关的数据。

EventBus.getDefault().post() 源码

还剩最后一个问题，我们在订阅者中定义的订阅方法（@Subscribe注解标注的）什么时候会被调用？

EventBus.getDefault().post()

从最上面的例子代码，我们也可以看到发布者代码只有一行，说明所有的类在任意地方都可以成为发布者发布事件，从而触发订阅者的订阅方法。其实，订阅者也可以是任意类，只是Android中常见的界面是Activity和Fragment，并且为了保证内存安全所以一定要在销毁的生命周期里调用注销api，只要任意类中在合理的位置进行注册和注销也是可以安全的使用EventBus的。

来看看post()源码吧：

public void post(Object event) {
    //1、
    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 {
            //2、
            while (!eventQueue.isEmpty()) {
                postSingleEvent(eventQueue.remove(0), postingState);
            }
        } finally {
            postingState.isPosting = false;
            postingState.isMainThread = false;
        }
    }
}

一目了然，将当前要出发的事件类型添加到队列中，在while循环中调用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);
    }

    可以忽略...
}

发现最后都会调用postSingleEventForEventType()方法：

private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
    CopyOnWriteArrayList<Subscription> subscriptions;
    synchronized (this) {
        //1、
        subscriptions = subscriptionsByEventType.get(eventClass);
    }
    if (subscriptions != null && !subscriptions.isEmpty()) {
        //2、
        for (Subscription subscription : subscriptions) {
            postingState.event = event;
            postingState.subscription = subscription;
            boolean aborted = false;
            try {
                //3、
                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容器获取到了所有跟当前订阅事件类型关联的Subscription对象集合，遍历调用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);
    }
}

该方法主要负责线程调度的过程，我们仅关注invokeSubscriber()即可：

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

通过我们之前在Subscription对象里保存的method对象和它的订阅者对象，以及当前事件类型对象以反射调用的形式触发了我们在订阅者中编写的订阅方法。

小结

通过以上的源码追踪，EventBus的内部实现我们已经有了一个清晰的认识，乘热打铁总结一下：

• register() -> 方法通过反射将当前订阅者中的订阅方法、订阅类型、订阅者进行了拆解存储到了两个容器中；
• unregister() -> 方法负责将当前的订阅者数据从那两个容器中移除；
• post() -> 方法在容器中找到跟当前订阅事件类型相关联的订阅方法、订阅者，并通过反射进行调用。