EventBus源码分析

EventBus简介

EventBus是一种基于Android的事件发布-订阅总线。

EventBus官网描述：

• simplifies the communication between components
  • decouples event senders and receivers
  • performs well with Activities, Fragments, and background threads
  • avoids complex and error-prone dependencies and life cycle issues
• makes your code simpler
• is fast
• is tiny (~50k jar)
• is proven in practice by apps with 100,000,000+ installs
• has advanced features like delivery threads, subscriber priorities, etc.

大致的意思是：EventBus能够简化各个组件之间的通信，解耦了事件发送与接收，让我们的代码更简洁快速，包50KB体积小，支持不同线程切换，具有优先级等特性。

EventBus源码分析

从函数构建开始分析：

• EventBus#getDefault()

//使用 volatile 关键字 使得线程可见
static volatile EventBus defaultInstance;

//双重验证加锁单例，确保唯一
public static EventBus getDefault() {
    if (defaultInstance == null) {
        synchronized (EventBus.class) {
            if (defaultInstance == null) {
                defaultInstance = new EventBus();
            }
        }
    }
    return defaultInstance;
}

事件的订阅：

• EventBus#getDefault()#register(Object)

//订阅事件的入口
public void register(Object subscriber) {
    Class<?> subscriberClass = subscriber.getClass();
    //内部通过反射方式，获取订阅者事件的方法
    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);
    //不为空则直接返回
    if (subscriberMethods != null) {
        return subscriberMethods;
    }
	//若不使用EventBusBuilder自己定义,ignoreGeneratedIndex 默认是 false,
    if (ignoreGeneratedIndex) {
        //直接通过放射获取
        subscriberMethods = findUsingReflection(subscriberClass);
    } else {
        //下面分析
        subscriberMethods = findUsingInfo(subscriberClass);
    }
    //如果订阅者中不存在@Subscribe注解的方法，则抛出异常
    if (subscriberMethods.isEmpty()) {
        throw new EventBusException("Subscriber " + subscriberClass
                + " and its super classes have no public methods with the @Subscribe annotation");
    } else {
        //放入缓存池中然后 return 订阅方法
        METHOD_CACHE.put(subscriberClass, subscriberMethods);
        return subscriberMethods;
    }
}

• findUsingInfo(subscriberClass)，这里有两种情况注意：

private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
    //初始化FindState
    FindState findState = prepareFindState();
    //将subscriberClass赋值
    findState.initForSubscriber(subscriberClass);
    //这里遍历有两种情况
    while (findState.clazz != null) {
        //1.通过APT插件生成的类中获取，避免使用反射，节约性能，这里暂不做分析，下篇研究使用该方式获取
        //需要导入apt:annotationProcessor 'org.greenrobot:eventbus-annotation-processor:3.1.1'
        findState.subscriberInfo = getSubscriberInfo(findState);
        if (findState.subscriberInfo != null) {
            //直接从apt生成的类中获取
            SubscriberMethod[] array = findState.subscriberInfo.getSubscriberMethods();
            for (SubscriberMethod subscriberMethod : array) {
                if (findState.checkAdd(subscriberMethod.method, subscriberMethod.eventType)) {
                    findState.subscriberMethods.add(subscriberMethod);
                }
            }
        } else {
            //2.直接通过反射的方式获取 注解@Subscribe定义的方法，下面继续分析该方法
            findUsingReflectionInSingleClass(findState);
        }
        //继续遍历父类
        findState.moveToSuperclass();
    }
    //通过上面的获取findState中的List<SubscriberMethod>，然后释放findState
    return getMethodsAndRelease(findState);
}

• 通过反射方式获取

//直接调用反射获取
private void findUsingReflectionInSingleClass(FindState findState) {
    Method[] methods;
    try {
        //通过反射获取所有方法
        methods = findState.clazz.getDeclaredMethods();
    } catch (Throwable th) {
     	//异常走这里
        methods = findState.clazz.getMethods();
        findState.skipSuperClasses = true;
    }
    //开始遍历所有方法
    for (Method method : methods) {
        int modifiers = method.getModifiers();
        //方法过滤掉不是public修饰，static,abstract,
        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];
                    //校验是否存在同名的方法，如果存在则替换，集合中添加eventType,与method
                    if (findState.checkAdd(method, eventType)) {
                        //获取方法所在的threadMode(线程模型)
                        ThreadMode threadMode = subscribeAnnotation.threadMode();
                        //将SubscriberMethod存储至findState中
                        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");
        }
    }
}

上述基本的获取 @Subscribe 注解分析到此就结束了，下面是分析获取到的method 及 eventType存储及订阅

• register()中的遍历集合中的SubscriberMethod，然后调用此方法subscribe()存储及订阅

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
    //获取事件类型
    Class<?> eventType = subscriberMethod.eventType;
    //构建 Subscription 便于管理以及事件发送
    Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
    //从集合中获取，若不存在，则创建添加
    CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
    if (subscriptions == null) {
        subscriptions = new CopyOnWriteArrayList<>();
        subscriptionsByEventType.put(eventType, subscriptions);
    } else {
    //对象已经调用过register(),重复调用,则抛出异常
        if (subscriptions.contains(newSubscription)) {
            throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "+ eventType);
        }
    }
    int size = subscriptions.size();
    //根据设置的priority向集合中相应的位置进行添加
    for (int i = 0; i <= size; i++) {
        if (i == size||subscriberMethod.priority>subscriptions.get(i).subscriberMethod.priority) {
            subscriptions.add(i, newSubscription);
            break;
        }
    }
	//typesBySubscriber用于管理订阅者注销
    List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
    if (subscribedEvents == null) {
        subscribedEvents = new ArrayList<>();
        typesBySubscriber.put(subscriber, subscribedEvents);
    }
    subscribedEvents.add(eventType);
	//判断该订阅方法是否为粘性事件，粘性事件发送使用，EventBus.postSticky(obj)
    if (subscriberMethod.sticky) {
        if (eventInheritance) {
            //遍历所有的粘性事件
            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 {
            //从stickyEvents集合中获取该事件，进行发送
            Object stickyEvent = stickyEvents.get(eventType);
            checkPostStickyEventToSubscription(newSubscription, stickyEvent);
        }
    }
}

注意：粘性事件存储在stickyEvents中，如果遍历的时候发现这个订阅方法为粘性，遍历stickyEvents集合，判断如果存在相同的eventType，则立即发送，到这里register()所有的源码分析就结束了。

事件的发送

EventBus通过post(obj)发送事件,期间包括线程的切换等

• post 发送事件event

public void post(Object event) {
    //获取当前线程中的PostingThreadState状态信息
    PostingThreadState postingState = currentPostingThreadState.get();
    List<Object> eventQueue = postingState.eventQueue;
    eventQueue.add(event);
	//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;
        }
    }
}

• postSingleEvent

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);
            //将获取到的event、postingState、class对象交付于postSingleEventForEventType()处理
            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));
        }
    }
}

• postSingleEventForEventType

private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
    CopyOnWriteArrayList<Subscription> subscriptions;
    synchronized (this) {
        //从集合中获取 register()存储的 Subscription
        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.subscription = null;
                postingState.canceled = false;
            }
            if (aborted) {
                break;
            }
        }
        return true;
    }
    return false;
}

• postToSubscription()

最终调用，通过threadMode进行相应的处理，通过反射、Handler.sendMessage()、提交任务至线程池执行

private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
    switch (subscription.subscriberMethod.threadMode) {
        case POSTING:
            //直接在调用Post所在的线程中调用该方法
            invokeSubscriber(subscription, event);
            break;
        case MAIN:
            //如果post()在主线程直接通过反射调用，否则通过handler的方式进行消息分发处理
            if (isMainThread) {
                invokeSubscriber(subscription, event);
            } else {
                mainThreadPoster.enqueue(subscription, event);
            }
            break;
        case MAIN_ORDERED:
            //Android下一通过handler的方式进行消息分发处理，不管是不是在UI线程
            if (mainThreadPoster != null) {
                mainThreadPoster.enqueue(subscription, event);
            } else {
                // HandlerPost创建失败，不在Android环境下直接调用反射
                invokeSubscriber(subscription, event);
            }
            break;
        case BACKGROUND:
            //如果post()在UI线程，则提交至线程池处理 调用run
            if (isMainThread) {
                backgroundPoster.enqueue(subscription, event);
            } else {
                //如果post()在子线程中直接通过反射调用
                invokeSubscriber(subscription, event);
            }
            break;
        case ASYNC:
            //使用ASYNC关键字，不管post()在什么线程中，都交付于线程池去处理
            asyncPoster.enqueue(subscription, event);
            break;
        default:
            throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
    }
}

EventBus主要是通过维护一个线程池与Handler，在不同的thredMode下进行事件的分发调用，这就所谓的不同threadMode下执行的环境也是不同

线程切换的两个类

• HandlerPoster

通过传入UI线程中的looper创建Handler使其在主线程中处理相应的Subscriber方法

public class HandlerPoster extends Handler implements Poster {
	//省略部分代码。。。。
    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) {
                //从PendingPostQueue取出PendingPost
                PendingPost pendingPost = queue.poll();
                if (pendingPost == null) {
                    synchronized (this) {
                        // 双重校验
                        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;
        }
    }
}

• BackgroundPoster

用于处理异步任务，通过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.getExecutorService().execute(this);
            }
        }
    }

    @Override
    public void run() {
        try {
            try {
                while (true) {
                    PendingPost pendingPost = queue.poll(1000);
                    if (pendingPost == null) {
                        synchronized (this) {
                            // 双重校验
                            pendingPost = queue.poll();
                            if (pendingPost == null) {
                                executorRunning = false;
                                return;
                            }
                        }
                    }
                    //继续调用EventBus.invokeSubscriber(), 在非UI线程中直接通过反射调用@Subscriber方法
                    eventBus.invokeSubscriber(pendingPost);
                }
            } catch (InterruptedException e) {
                eventBus.getLogger().log(Level.WARNING, Thread.currentThread().getName() + " was interruppted", e);
            }
        } finally {
            executorRunning = false;
        }
    }

}

AsyncPoster与BackgroundPoster类似，都是通过提交run至线程池处理，公用一个线程池

订阅注销操作

public synchronized void unregister(Object subscriber) {
    //register中提到过这个类，专门用户注销订阅者集合
    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());
    }
}

到这里EventBus主要源码分析到此就结束了，感谢收看。