EventBus源码分析

差不多两年没写博客了，最近想着要找工作了，打算复习下一些常用的开源库，也是这篇博客的由来～

EventBus使用非常简单参考：github
再贴一张官网的图
在这里插入图片描述

一、示例代码

示例代码是为了便于理解后面注解处理器生成代码的处理流程

public class TestRunnerActivity extends Activity {

    private EventBus eventBus;

    @Override
    public void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_runtests);
        eventBus = new EventBus();
        eventBus.register(this);
    }

    @Subscribe(threadMode = ThreadMode.MAIN)
    public void onEventMainThread(TestFinishedEvent event) {

    }

    @Subscribe(threadMode = ThreadMode.BACKGROUND)
    public void onEventBackgroundThread(TestEvent event) {
    }

    public void onDestroy() {
        eventBus.unregister(this);
        super.onDestroy();
    }
}

二、EventBus#register流程

调用register进行订阅 ,参数this代表订阅者

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

首先调用了findSubscriberMethods获取了当前订阅者类里的所有订阅方法。

List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
        //如果缓存有，直接返回。 METHOD_CACHE类型是Map<Class<?>, List<SubscriberMethod>>
        List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
        if (subscriberMethods != null) {
            return subscriberMethods;
        }
        //ignoreGeneratedIndex默认是false
        if (ignoreGeneratedIndex) {
            //方式1:通过反射获取到所有的订阅方法
            subscriberMethods = findUsingReflection(subscriberClass);
        } else {
            //方式2:通过注解处理器生成的代码来获取所有的订阅方法
            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;
        }
    }

可以看到有两种方式获取订阅方法，一种是利用反射，另一种是利用注解处理器生成的代码。很多开源库都利用了注解生成器，通过它在编译时生成代码可以很好的帮助我们减少反射代码的调用。

先看看如何通过反射获取到所有的订阅方法

private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
        //从对象缓存池获取一个FindState，这只是一个工具对象
        FindState findState = prepareFindState();
   			//给findState成员变量subscriberClass 、class赋值为subscriberClass
        findState.initForSubscriber(subscriberClass);
        while (findState.clazz != null) {
          	//通过反射去找
            findUsingReflectionInSingleClass(findState);
            //继续找父类中的所有方法
            findState.moveToSuperclass();
        }
        //返回找到的List<SubscriberMethod>，回收FindState
        return getMethodsAndRelease(findState);
    }

反射的核心逻辑在findUsingReflectionInSingleClass方法,该方法通过反射查找class的所有方法，然后遍历这些方法找到带@Subscribe的方法，然后将查询的结果保存到findState.subscriberMethods中

private void findUsingReflectionInSingleClass(FindState findState) {
        Method[] methods;
        try {
            //getDeclaredMethod()获取的是类自身声明的所有方法，包含public、protected和private方法
            methods = findState.clazz.getDeclaredMethods();
        } catch (Throwable th) {
            ...
        }
        //遍历所有方法，找到带Subscribe注解的方法 比如：@Subscribe(threadMode = ThreadMode.MAIN)
        for (Method method : methods) {
            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];
                        //检查method&eventType对于的订阅方法是否已经添加过了
                        if (findState.checkAdd(method, eventType)) {
                            //获取方法的线程模式
                            ThreadMode threadMode = subscribeAnnotation.threadMode();
                            //最终将找到的结果 封装到SubscriberMethod里面，然后保存到findState.subscriberMethods中
                            findState.subscriberMethods.add(new SubscriberMethod(method, eventType, threadMode,
                                    subscribeAnnotation.priority(), subscribeAnnotation.sticky()));
                        }
                    }
                } 
            } 
        }
    }

前面的findUsingReflection方法先通过反射找到订阅类里的所有订阅方法，然后保存到findState.subscriberMethods中，接着调用getMethodsAndRelease(findState),将
List< SubscriberMethod>返回，然后回收FindState到对象池

private List<SubscriberMethod> getMethodsAndRelease(FindState findState) {
        List<SubscriberMethod> subscriberMethods = new ArrayList<>(findState.subscriberMethods);
        findState.recycle();
        synchronized (FIND_STATE_POOL) {
            for (int i = 0; i < POOL_SIZE; i++) {
                if (FIND_STATE_POOL[i] == null) {
                    FIND_STATE_POOL[i] = findState;
                    break;
                }
            }
        }
        return subscriberMethods;
    }

Ok，现在我们已经通过反射获取到订阅类中的所有订阅方法了，并且将它们封装到List< SubscriberMethod>中了，我们先不管注解处理器获取List< SubscriberMethod>的逻辑，继续看register方法，获取到List< SubscriberMethod>后开始遍历List< SubscriberMethod>，对每个SubscriberMethod调用subscribe

subscribe主要干了两件事

1、将订阅方法和订阅者封装到Subscription中，然后为每个事件类型创建一个CopyOnWriteArrayList列表subscriptions，并将Subscription插入到subscriptions中。事件类型就是我们post的事件的class对象

2、对粘性事件进行处理

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
        //获取事件类型
        Class<?> eventType = subscriberMethod.eventType;
        //将订阅者和订阅方法封装到Subscription中
        Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
        //通过事件类型找到Subscription集合
        CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
        if (subscriptions == null) {
            //如果还没有订阅类 订阅了该事件的方法，就创建一个list,把当前的Subscription放进去
            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中
        subscribedEvents.add(eventType);

  			//下面这个逻辑可以先不看 后面可以结合postSticky流程看
        //如果订阅方法支持粘性事件
        if (subscriberMethod.sticky) {
            if (eventInheritance) {
                Set<Map.Entry<Class<?>, Object>> entries = stickyEvents.entrySet();
                //遍历stickyEvents map  这个map保存了粘性事件
                for (Map.Entry<Class<?>, Object> entry : entries) {
                    Class<?> candidateEventType = entry.getKey();
                    //如果当前订阅方法订阅的事件列席在stickyEvents中已经有了，就说明之前某个地方发送过该粘性事件
                    if (eventType.isAssignableFrom(candidateEventType)) {
                        Object stickyEvent = entry.getValue();
                        //检查发送粘性事件
                        checkPostStickyEventToSubscription(newSubscription, stickyEvent);
                    }
                }
            } else {
                Object stickyEvent = stickyEvents.get(eventType);
                //检查发送粘性事件
                checkPostStickyEventToSubscription(newSubscription, stickyEvent);
            }
        }
    }

走到这里我们的注册逻辑基本就结束了，注册的结果就是得到了一个subscriptionsByEventType，subscriptionsByEventType是一个map key是事件类型，value是该事件的所有订阅(Subscription)。

注册的流程走完了再来看看如何通过注解处理器生成的代码获取SubscriberMethod列表。
下面这个类MyEventBusIndex是注解处理器生成的

public class MyEventBusIndex implements SubscriberInfoIndex {
    private static final Map<Class<?>, SubscriberInfo> SUBSCRIBER_INDEX;

    static {
        key 订阅者类  value 订阅信息
        SUBSCRIBER_INDEX = new HashMap<Class<?>, SubscriberInfo>();
     	//...
        //将订阅信息保存到SUBSCRIBER_INDEX
        putIndex(new SimpleSubscriberInfo(TestRunnerActivity.class, true, new SubscriberMethodInfo[] {
            new SubscriberMethodInfo("onEventMainThread", TestFinishedEvent.class, ThreadMode.MAIN),
            new SubscriberMethodInfo("onEventBackgroundThread", TestEvent.class, ThreadMode.BACKGROUND),
        }));

    }

    private static void putIndex(SubscriberInfo info) {
      	//保存所有订阅类的订阅信息
        SUBSCRIBER_INDEX.put(info.getSubscriberClass(), info);
    }

    @Override
    public SubscriberInfo getSubscriberInfo(Class<?> subscriberClass) {
        SubscriberInfo info = SUBSCRIBER_INDEX.get(subscriberClass);
        if (info != null) {
            return info;
        } else {
            return null;
        }
    }
}

SimpleSubscriberInfo定义如下：

public class SimpleSubscriberInfo extends AbstractSubscriberInfo {

    private final SubscriberMethodInfo[] methodInfos;

    public SimpleSubscriberInfo(Class subscriberClass, boolean shouldCheckSuperclass, SubscriberMethodInfo[] methodInfos) {
        super(subscriberClass, null, shouldCheckSuperclass);
        this.methodInfos = methodInfos;
    }

    @Override
    public synchronized SubscriberMethod[] getSubscriberMethods() {
        int length = methodInfos.length;
        SubscriberMethod[] methods = new SubscriberMethod[length];
        for (int i = 0; i < length; i++) {
            SubscriberMethodInfo info = methodInfos[i];
            methods[i] = createSubscriberMethod(info.methodName, info.eventType, info.threadMode,
                    info.priority, info.sticky);
        }
        return methods;
    }
}

MyEventBusIndex类对象会被保存到EventBus的SubscriberMethodFinder中subscriberInfoIndexes成员列表中

private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
        FindState findState = prepareFindState();
        findState.initForSubscriber(subscriberClass);
        while (findState.clazz != null) {
            //找到index类中的订阅信息 ,其实就是调用上面MyEventBusIndex的getSubscriberInfo获取订阅类的SubscriberInfo
            findState.subscriberInfo = getSubscriberInfo(findState);
            if (findState.subscriberInfo != null) {
                //获取SimpleSubscriberInfo的getSubscriberMethods方法获取订阅信息
                SubscriberMethod[] array = findState.subscriberInfo.getSubscriberMethods();
                for (SubscriberMethod subscriberMethod : array) {
                    //如果没有添加过，就添加订阅方法
                    if (findState.checkAdd(subscriberMethod.method, subscriberMethod.eventType)) {
                        findState.subscriberMethods.add(subscriberMethod);
                    }
                }
            } else {
                //如果index里没有，就通过反射来查找订阅类中的订阅方法
                findUsingReflectionInSingleClass(findState);
            }
            //继续找父类的
            findState.moveToSuperclass();
        }
        //返回订阅方法列表
        return getMethodsAndRelease(findState);
    }

ok到现在为止，注册的逻辑终于结束了。

三、EventBus#unregister流程

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 {
            logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
        }
    }

最终是通过事件类型来解除该订阅者的所有注册

private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
        //获取所有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) {
                    //将active 状态改为 false
                    subscription.active = false;
                    //移除订阅关系
                    subscriptions.remove(i);
                    i--;
                    size--;
                }
            }
        }
    }

四、EventBus#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;
        try {
            while (!eventQueue.isEmpty()) {
                //遍历所有的事件，进行发送
                postSingleEvent(eventQueue.remove(0), postingState);
            }
        } finally {
            postingState.isPosting = false;
            postingState.isMainThread = false;
        }
    }
}

然后调用postSingleEvent进行单个event的发送

private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
        Class<?> eventClass = event.getClass();
        boolean subscriptionFound = false;
        //是否考虑父类的事件类型
        if (eventInheritance) {
            //遍历获取当前的event类以及其父类的类型
            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) {
            //获取所有该事件类型的订阅对象
            subscriptions = subscriptionsByEventType.get(eventClass);
        }
        if (subscriptions != null && !subscriptions.isEmpty()) {
            for (Subscription subscription : subscriptions) {
                postingState.event = event;
                postingState.subscription = subscription;
                boolean aborted;
                try {
                    //遍历所有的Subscription 发送事件
                    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这个map 在分析register流程的时候分析过它的key是事件类型，value是该事件的所有订阅(Subscription)。

接着遍历所有的Subscription调用postToSubscription方法

private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
        switch (subscription.subscriberMethod.threadMode) {
            case POSTING://发送方在哪里线程，订阅方法就在哪个方法被调用
                //直接调用invokeSubscriber,关于invokeSubscriber后面会说
                invokeSubscriber(subscription, event);
                break;
            case MAIN://只在主线程调用订阅方法
                if (isMainThread) {
                    //如果当前是在主线程直接调用invokeSubscriber
                    invokeSubscriber(subscription, event);
                } else {
                    //如果不在主线程，利用handler机制切到主线程再调用invokeSubscriber
                    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) {
                    //如果在主线程就切到子线程后调用invokeSubscriber
                    backgroundPoster.enqueue(subscription, event);
                } else {
                    //如果在子线程直接调用invokeSubscriber
                    invokeSubscriber(subscription, event);
                }
                break;
            case ASYNC:
                //利用线程池执行invokeSubscriber方法
                asyncPoster.enqueue(subscription, event);
                break;
            default:
                throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
        }
    }

不管是在子线程还是在主线程最终都是通过invokeSubscriber方法，利用反射调用订阅对象的订阅方法

void invokeSubscriber(Subscription subscription, Object event) {
        subscription.subscriberMethod.method.invoke(subscription.subscriber, event);
    }

五、EventBus#postSticky粘性事件发送流程

public void postSticky(Object event) {
    //粘性事件会被保存到stickyEvents中
    synchronized (stickyEvents) {
        stickyEvents.put(event.getClass(), event);
    }
    //正常发送事件
    post(event);
}

和普通事件不同之处在于粘性事件会被保存到stickyEvents中，这样即使在我们订阅之前已经发送了某个粘性事件，在我们订阅的时候也能收到，还记得订阅流程中有这样几行代码吗？

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
        //...
        //如果订阅方法支持粘性事件
        if (subscriberMethod.sticky) {
            if (eventInheritance) {
                Set<Map.Entry<Class<?>, Object>> entries = stickyEvents.entrySet();
                //遍历stickyEvents map  这个map保存了粘性事件
                for (Map.Entry<Class<?>, Object> entry : entries) {
                    Class<?> candidateEventType = entry.getKey();
                    //如果当前订阅方法订阅的事件列席在stickyEvents中已经有了，就说明之前某个地方发送过该粘性事件
                    if (eventType.isAssignableFrom(candidateEventType)) {
                        Object stickyEvent = entry.getValue();
                        //检查发送粘性事件
                        checkPostStickyEventToSubscription(newSubscription, stickyEvent);
                    }
                }
            } else {
                Object stickyEvent = stickyEvents.get(eventType);
                //检查发送粘性事件
                checkPostStickyEventToSubscription(newSubscription, stickyEvent);
            }
        }
    }

到这里EventBus的源码基本就分析完了，是不是很简单？