EventBus的使用

在onCreate()或者onStart()里面注册

EventBus.getDefault().register(this);

在onStop()或者onDestory()里面反注册

EventBus.getDefault().unregister(this);

通过post()或者postSticky发送一个消息或者粘性消息

MessageEvent messageEvent = new MessageEvent();
messageEvent.setAge(15);
messageEvent.setName("小明");
EventBus.getDefault().post(messageEvent);

EventBus.getDefault().postSticky(messageEvent);

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

public EventBus() {
        this(DEFAULT_BUILDER);
    }

可以发现EventBus的构造函数，为public,一般单例模式其构造函数为private。所以EventBus可以创建多个实例。并且多个实例的发送的事件是互不干扰的。

public void register(Object subscriber) {
        Class<?> subscriberClass = subscriber.getClass();
        List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
        synchronized (this) {
            for (SubscriberMethod subscriberMethod : subscriberMethods) {
                subscribe(subscriber, subscriberMethod);
            }
        }
    }

可以看到往往我们注册时传入的是this,所以这里首先调用getClass(),获得的就是MainActivity.class（举例），获得完这个class对象后，会调用findSubscriberMethods方法来找到该Class中所有的订阅方法。

List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
        List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
        if (subscriberMethods != null) {
            return subscriberMethods;
        }

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

如果缓存中有，则直接返回。如果没有，则有两种方法去找。一种通过反射，另外一种通过通过编译期注解来实现映射表的构建。ignoreGeneratedIndex表示是否忽略注解器生成的索引类。默认为false，可以通过builder来判断。

findUsingInfo（）

private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
        FindState findState = prepareFindState();
        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);
            }
            findState.moveToSuperclass();
        }
        return getMethodsAndRelease(findState);
    }

先通过prepareFindState()创建一个FindState对象，然后将传入的class对象赋值到findState中。

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

重点关注subscriberInfo，在这个方法的时候设置为null。

然后循环，findState.clazz肯定不为null，所以走下面，

private SubscriberInfo getSubscriberInfo(FindState findState) {
        if (findState.subscriberInfo != null && findState.subscriberInfo.getSuperSubscriberInfo() != null) {
            SubscriberInfo superclassInfo = findState.subscriberInfo.getSuperSubscriberInfo();
            if (findState.clazz == superclassInfo.getSubscriberClass()) {
                return superclassInfo;
            }
        }
        if (subscriberInfoIndexes != null) {
            for (SubscriberInfoIndex index : subscriberInfoIndexes) {
                SubscriberInfo info = index.getSubscriberInfo(findState.clazz);
                if (info != null) {
                    return info;
                }
            }
        }
        return null;
    }

上面说了subscriberInfo == null，所以不会走第一个if,而我们也没有试用索引，所以第二个也不会走。所以返回null。

所以在findUsingInfo（）中直接走else。

findUsingReflectionInSingleClass(findState);

private void findUsingReflectionInSingleClass(FindState findState) {
        Method[] methods;
        try {
            methods = findState.clazz.getDeclaredMethods();//1
        } catch (Throwable th) {
            ......
        }
        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) {
                        //获得 Event 事件的 Class 对象
                        Class<?> eventType = parameterTypes[0];
                        //检查是否需要保存到 findState 的 anyMethodByEventType 中, 返回 true 表示添加成功
                        if (findState.checkAdd(method, eventType)) {
                        //获得这个方法注解的 ThreadMode 线程模式, 根据不同的模式进行不同的线程调度
                            ThreadMode threadMode = subscribeAnnotation.threadMode();
                            findState.subscriberMethods.add(new SubscriberMethod(method, eventType, threadMode,
                                    subscribeAnnotation.priority(), subscribeAnnotation.sticky()));
                        }
                    }
                } else if (strictMethodVerification && method.isAnnotationPresent(Subscribe.class)) {
                    
                }
            } else if (strictMethodVerification && 
            }
        }
    }

在这里直接用反射的方式。

1处调用的是getDeclaredMethods()，而不是getMethods()。这样会提升性能。因为getDeclaredMethods()本类所有的方法，而getMethods()是获取本类以及父类所有的public修饰的方法。因为在findUsingInfo（）中的循环里调用了moveToSuperclass（），这个方法的作用就是向上寻找父类。然后进行循环，如果采用getMethods()就会重复，本来反射就是耗性能的。

然后返回getMethodsAndRelease(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;
    }

就是将findState放入到FIND_STATE_POOL池中，方便复用。

上面讲的是通过编译期注解来实现映射表的构建，而还有另外一种方法通过反射来实现。

findUsingReflection

private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
        FindState findState = prepareFindState();
        findState.initForSubscriber(subscriberClass);
        while (findState.clazz != null) {
            findUsingReflectionInSingleClass(findState);
            findState.moveToSuperclass();
        }
        return getMethodsAndRelease(findState);
    }

里面调用的还是findUsingReflectionInSingleClass（）。

然后回到订阅方法：

subscribe

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.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.add(eventType);

        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 {
                Object stickyEvent = stickyEvents.get(eventType);
                checkPostStickyEventToSubscription(newSubscription, stickyEvent);
            }
        }
    }

在订阅方法中有两个map需要注意，一个是subscriptionsByEventType，这个map的key是Event事件，value则是订阅者与订阅方法封装而成的Subscription。另一个则是typesBySubscriber，这个map的key是订阅者，value是Event事件。

到这里 EventBus 的注册流程分析完了. 总结后大致流程如下:

根据 EventBus.getDefault().register(this) 中的这个 this, 作为订阅者.

根据订阅者获取订阅者内部的所有订阅方法.

接着遍历订阅者内部的所有订阅方法, 进行订阅.

订阅的时候会先判断是否订阅过这个事件.

按照优先级将订阅记录加入到 subscriptionsByEventType 的 value 的 list 中

将当前事件类型加入到订阅者对应的订阅事件列表中 (typesBySubscriber 的 value 的 list 中)

是否是粘性事件.

调用 checkPostStickyEventToSubscription 分发事件.

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

这里的currentPostingThreadState是一个ThreadLocal类型，既然选择TreadLocal类型，那么特点当然就是线程单例，不同的线程之间互相不干扰，因为我们知道EventBus是支持多线程之间的事件传递的。

然后拿到事件队列，将事件添加到队列中，如果不是正在发送，循环从队首取消息，发送消息。

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

eventInheritance默认是false，当eventInheritance为true时，会调用lookupAllEventTypes方法，返回一个List集合。这个方法的所用就是用一个List保存该Event类型自己本身，和其继承的所有父类，和实现的所有接口。
所以当得到这个List集合后，会遍历这个集合，调用postSingleEventForEventType发送事件，所以这里可以看出默认情况下不要随意将Event继承或者实现接口，当发送该Event的时候，都会一并发出。

然后调用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 {
                    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;
    }

从前面讲的第一个map得到所有的订阅方法。然后遍历这个List调用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);
        }
    }

我们知道有五种线程模型：

POSTING：哪个线程发布就哪个线程处理。

MAIN：如果发布者现在是在主线程, 那么直接在主线程通过反射来处理事件，如果发布者不是在主线程, 就把当前事件加入到主线程的消息循环队列.

MAIN_ORDERED：在订阅的时候指定主线程执行,但是要先入队，

BACKGROUND：如果发布者是在主线程, 就把当前事件加入到后台线程消息循环队列.发布者不是在主线程, 那么就直接在发布者所在的线程反射处理事件.

ASYNC：将当前事件与订阅记录加入到异步消息队列后通过线程池为每个事件单独开辟一个线程进行执行.

首先我们来看三个poster。

mainThreadPoster（HandlerPoster）

public void enqueue(Subscription subscription, Object event) {
        //将subscription和event封装成一个PendingPost
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        synchronized (this) {
            //将PendingPost加入到队列中。
            queue.enqueue(pendingPost);
            if (!handlerActive) {
                handlerActive = true;
                //通过handler发送消息
                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 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;
                if (timeInMethod >= maxMillisInsideHandleMessage) {
                    if (!sendMessage(obtainMessage())) {
                        throw new EventBusException("Could not send handler message");
                    }
                    rescheduled = true;
                    return;
                }
            }
        } finally {
            handlerActive = rescheduled;
        }
    }

BackgroundPoster

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) {
                            // 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;
        }
    }

通过 enqueue() 方法可以看到如果 BackgroundPoster这个 Runnable 正在被线程池执行, 这时候 executorRunning==true, 那么在 executorRunning==true 的情况下发布的事件只会入队, 而不会再次去调用线程池的 execute() 方法.

这样的话在 run() 方法中就通过死循环遍历处理队列中的事件. 全部处理完成后就退出死循环. 然后设置 executorRunning 为 false, 表示没有正在运行中的线程了. 此后再发布事件才会在线程池中开辟一个新线程

AsyncPoster

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

异步模式会将订阅记录加入到异步消息队列后会通过线程池为每个事件单独开辟一个线程进行执行. 细看它的 enqueue() 方法. 它会为每一个添加的任务都在线程池中开辟一个新的线程执行. 并发度更高.

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

从typesBySubscribermap中取出当前订阅者的订阅的所有事件集合List，然后再遍历这个List,调用unsubscribeByEventType方法进行取消订阅，其实很容易联想到这个方法肯定是再到另一个重要的Map中查找订阅了这个Event的订阅者中移除当前Activity的订阅者。

遍历该Activity中订阅的所有事件，然后通过Event到subscriptionsByEventType中查找订阅了该Event的List<Subscription>集合，然后找到当前这个订阅者（注意这里使用的是，来判断两个Object对象是否相等，意味着是内存地址相同），找到后移除。至此注销流程也到此结束。