EventBus源码分析

问题

• 使用限制
• 事件发送以及接收原理
• 设计模式

分析

EventBus#getDefault()

 // 单例模式 双重检验
    public static EventBus getDefault() {
        EventBus instance = defaultInstance;
        if (instance == null) {
            synchronized (EventBus.class) {
                instance = EventBus.defaultInstance;
                if (instance == null) {
                    instance = EventBus.defaultInstance = new EventBus();
                }
            }
        }
        return instance;
    }

 // 构造方法不是私有的 则表明可以存在多个EventBus
    public EventBus() {
        this(DEFAULT_BUILDER);
    }

EventBus#register()

public void register(Object subscriber) {
        Class<?> subscriberClass = subscriber.getClass();
        // TODO: 2018/5/24 根据注册的类寻找类中绑定的方法
        List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
        synchronized (this) {
            for (SubscriberMethod subscriberMethod : subscriberMethods) {
                subscribe(subscriber, subscriberMethod);
            }
        }
    }

 List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
        // 缓存中的方法订阅列表 
        List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
        // 如果缓存中有对应class的订阅方法列表 那么直接返回此列表 
        if (subscriberMethods != null) {
            return subscriberMethods;
        }

        // 此值一般为false 
        if (ignoreGeneratedIndex) {
            subscriberMethods = findUsingReflection(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;
        }
    }

private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
        // 准备一个FindState 保存订阅者类的消息 
        FindState findState = prepareFindState();

        // 根据subscriberClass类对FindState进行初始化
        findState.initForSubscriber(subscriberClass);
        
        while (findState.clazz != null) {
            // 获取订阅者的信息 
            findState.subscriberInfo = getSubscriberInfo(findState);
            if (findState.subscriberInfo != null) {
                SubscriberMethod[] array = findState.subscriberInfo.getSubscriberMethods();
                for (SubscriberMethod subscriberMethod : array) {
                    if (findState.checkAdd(subscriberMethod.method, subscriberMethod.eventType)) {
                        findState.subscriberMethods.add(subscriberMethod);
                    }
                }
            } else {
                findUsingReflectionInSingleClass(findState);
            }
            // 到父类继续查找订阅者的信息 如果超类是系统的类返回null也就跳出类while循环
            findState.moveToSuperclass();
        }
        // 获取到订阅方法列表并且回收findState
        return getMethodsAndRelease(findState);
    }

// FindState准备此类时使用类缓存 享元模式
    private FindState prepareFindState() {
        synchronized (FIND_STATE_POOL) {
            for (int i = 0; i < POOL_SIZE; i++) {
                FindState state = FIND_STATE_POOL[i];
                if (state != null) {
                    FIND_STATE_POOL[i] = null;
                    return state;
                }
            }
        }
        return new FindState();
    }

private void findUsingReflectionInSingleClass(FindState findState) {
        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;
        }
        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作为注解的方法处理
                    Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
                    if (subscribeAnnotation != null) {
                        Class<?> eventType = parameterTypes[0];
                        if (findState.checkAdd(method, eventType)) {
                            ThreadMode threadMode = subscribeAnnotation.threadMode();
                            // 将方法名，threadMode，优先级，是否为sticky方法封装为SubscriberMethod对象，添加到subscriberMethods列表中 
                            findState.subscriberMethods.add(new SubscriberMethod(method, eventType, threadMode,
                                    subscribeAnnotation.priority(), subscribeAnnotation.sticky()));
                        }
                    }
                } 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);
                }
            } 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");
            }
        }
    }

SubscriberMethodFinder#getMethodsAndRelease

private List<SubscriberMethod> getMethodsAndRelease(FindState findState) {
        List<SubscriberMethod> subscriberMethods = new ArrayList<>(findState.subscriberMethods);
        // 回收findState
        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;
                }
            }
        }
        // 最终返回这个subscriberMethods列表
        return subscriberMethods;
    }

EventBus#subscribe()

// 此方法的核心 将所有含@Subscribe注解的订阅方法最终保存在subscriptionsByEventType中
    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是一个map key是eventType value是subscriptions 
            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++) {
            // 判断订阅的方法的优先级 优先级高的排在subscriptions的前面 
            if (i == size || subscriberMethod.priority > subscriptions.get(i).subscriberMethod.priority) {
                // newSubscription包含newSubscription和newSubscription
                subscriptions.add(i, newSubscription);
                break;
            }
        }

        List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
        if (subscribedEvents == null) {
            subscribedEvents = new ArrayList<>();
            // subscriptions 所有的eventType列表 在isRegister()中使用 判断这个subscriber是否已经被注册过
            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);
            }
        }
    }

EventBus#checkPostStickyEventToSubscription()

  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, isMainThread());
        }
    }

EventBus#post()

public void post(Object event) {
        // currentPostingThreadState是ThreadLocal类型 存储了PostingThreadState PostingThreadState中包含了一个eventQueue和其他一些标志位
        PostingThreadState postingState = currentPostingThreadState.get();
        List<Object> eventQueue = postingState.eventQueue;
        // 将event保存在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;
            }
        }
    }

final static class PostingThreadState {
        final List<Object> eventQueue = new ArrayList<>();
        // 防止方法多次调用 
        boolean isPosting;
        // 判断是否是ui线程 
        boolean isMainThread;
        Subscription subscription;
        Object event;
        boolean canceled;
    }

EventBus#postSingleEvent()

private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
        // 获取事件的class 所有事件的class对应的订阅者列表在register的时候已经保存了
        Class<?> eventClass = event.getClass();
        // 根据事件的Class找到订阅者的标志状态
        boolean subscriptionFound = false;
        // eventInheritance 此标志简单的讲就是是否考虑事件的继承性 如果是true则需要找到事件所有父类和所有实现的接口
        if (eventInheritance) {
            // 取出Event的父类和接口的class列表
            List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
            int countTypes = eventTypes.size();
            for (int h = 0; h < countTypes; h++) {
                Class<?> clazz = eventTypes.get(h);
                // 基础要复习了 此处相当于 subscriptionFound=subscriptionFound|postSingleEventForEventType(event, postingState, clazz)
                subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
            }
        } else {
            subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
        }
        if (!subscriptionFound) {
            if (logNoSubscriberMessages) {
                logger.log(Level.FINE, "No subscribers registered for event " + eventClass);
            }
            if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
                    eventClass != SubscriberExceptionEvent.class) {
                post(new NoSubscriberEvent(this, event));
            }
        }
    }

EventBus#postToSubscription()

 // 此方法中区分处理事件的线程
    private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
        switch (subscription.subscriberMethod.threadMode) {
            case POSTING:
                // 订阅者方法将在发布事件所在的线程中被调用 默认的线程模式
                invokeSubscriber(subscription, event);
                break;
            case MAIN:
                // 如果是主线程那就直接调用 否则加入队列中 后续通过handler去发送一个消息 具体逻辑在HandlerPoster中
                if (isMainThread) {
                    invokeSubscriber(subscription, event);
                } else {
                    mainThreadPoster.enqueue(subscription, event);
                }
                break;
            case MAIN_ORDERED:
                // 此种模式下事件处理会按一定的顺序执行 先发布的先处理完；区别与MAIN的模式下 处理事件的时候发布另一个事件 那么第二个事件会先于第一个事件处理
                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) {
                    // 是ui线程放入后台的队列 通过线程池调用
                    backgroundPoster.enqueue(subscription, event);
                } else {
                    // 不是ui线程直接反射调用
                    invokeSubscriber(subscription, event);
                }
                break;
            case ASYNC:
                // 与BACKGROUND的逻辑类似，将任务加入到后台的一个队列，最终由Eventbus中的一个线程池去调用
                // 与BACKGROUND的区别是 ASYNC是异步 而BACKGROUND是同步
                asyncPoster.enqueue(subscription, event);
                break;
            default:
                throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
        }
    }

总结

• 使用限制
  绑定方法的命名以及线程的限制

• 事件发送接收原理
  核心：注解+反射调用注册的方法

  
  流程.png
• 设计模式
  观察者 享元 单例