进击的巨人——EventBus源码解析

“If I have seen further it is by standing on ye shoulders of Giants。” —— Newton

开篇简介：巨人系列，是打算为读者带来我所阅读的源码做一些分析与讲解，详细的娓娓道来可能对现在的我来说，时间与水平上都不足够，只能为大家带来粗浅的东西。但愿能对读者提供些许帮助。

事件总线设计框架——EventBus

提起事件总线设计却不得不提GoF的观察者模式，可以说前者吸取了后者的精髓——订阅/发布机制

观察者模式结构示意图：

事件总线设计结构示意图：

在此我归档出几个显而易见的不同点

区别	观察者模式	事件总线
关系类型	1 –> N	N –> N
包含对象	主体对象（1）观察者对象（N）	事件总线（一般为1）事件（N）事件监听器（N）
信息传递方式	直接传递	包裹在事件中

Ok ！概念总是来唬人的，就像我考了三次都没过的党课一样鸡肋。不过当你知道如何实践之后，概念就是你的总结

接下来开始大致的介绍下EventBus一些基本的类，便于对代码的理解。
以下是代码关联结构图：
（声明：以下图多为来自codeKK的EventBus讲解，比我讲解的更加全面，不过没有源码说明，想看的朋友不妨看完我的，再移步不迟）

从上图我们可以看到3种Poster，他们对应着不同的ThreadMode

模式	描诉
onEvent	如果使用 onEvent作为订阅函数，那么该事件在哪个线程发布出来的，onEvent就会在这个线程中运行，也就是说发布事件和接收事件线程在同一个线程。使用这个方法时，在 onEvent方法中不能执行耗时操作，如果执行耗时操作容易导致事件分发延迟
onEventMainThread	如果使用onEventMainThread作为订阅函数，那么不论事件是在哪个线程中发布出来的，onEventMainThread都会在UI线程中执行，接收事件就会在UI线程中运行，这个在Android中是非常有用的，因为在Android中只能在UI线程中跟新UI，所以在onEvnetMainThread 方法中是不能执行耗时操作的
onEventBackground	如果使用onEventBackgrond作为订阅函数，那么如果事件是在UI线程中发布出来的，那onEventBackground就会在子线程中运行，如果事件本来就是子线程中发布出来的，那么onEventBackground函数直接在该子线程中执行。
onEventAsync	使用这个函数作为订阅函数，那么无论事件在哪个线程发布，都会创建新的子线程在执行 onEventAsync.

以下是下面常出现的名词

概念
事件	传递的信息
订阅者	需要监听信息的东西？这样描述吧…
事件监听器	接收到事件后，订阅者要做的事（方法）
订阅信息	订阅者+事件监听器

给点时间，大概五分钟吧，了解这个结构图、不同的模式和基本名词。大家就已经基本了解EventBus的代码结构，除了一些边角的类如异常类，配置类，工具类等,与主体功能无关。
初步了解大致功能运行流程，之后再深入阅读，假如时间足够。XD。

接下来，马上就要看EventBus的源码了，在看源码前，大家可以思考下，对EventBus那些东西最好奇？抱着问题去看，以期在源码中找到解答。
于我而言，我想知道这些东西：

EventBus是如何在通知其他Activity或Fragment刷新UI线程的？
EventBus使用的时候为什么要注册和反注册，为什么传 Context 呢？
EventBus 如何实现多种线程模式的？
EventBus 使用的时候，为什么要要求取方法名一致？

答案就在代码中~

How time flies！
五分钟回来。开始讲诉我们的EventBus的基本流程啦~

first step： register
1. getDefault()

Only a singleton ~ 主要做的还是一些基本的变量声明之类的，不细说。下面列出构造函数

 EventBus(EventBusBuilder builder) {//关联着所有的<事件，订阅信息>subscriptionsByEventType = new HashMap<Class<?>, CopyOnWriteArrayList<Subscription>>();//关联着所有的<订阅者，事件>typesBySubscriber = new HashMap<Object, List<Class<?>>>();//储存stick事件stickyEvents = new ConcurrentHashMap<Class<?>, Object>();//对应着onEventMainThread的postermainThreadPoster = new HandlerPoster(this, Looper.getMainLooper(), 10);//对应着onEventBackgroundbackgroundPoster = new BackgroundPoster(this);//对应着onEventAsyncasyncPoster = new AsyncPoster(this);//提供查找所有的方法的对象subscriberMethodFinder = new SubscriberMethodFinder(builder.skipMethodVerificationForClasses);logSubscriberExceptions = builder.logSubscriberExceptions;logNoSubscriberMessages = builder.logNoSubscriberMessages;sendSubscriberExceptionEvent = builder.sendSubscriberExceptionEvent;sendNoSubscriberEvent = builder.sendNoSubscriberEvent;throwSubscriberException = builder.throwSubscriberException;//是否继承eventInheritance = builder.eventInheritance;//线程池executorService = builder.executorService;}

2.register(this)
EventBus 中有5个register * 的方法，不过他们都是使用了同一个 register：

 /**** @param subscriber    任意class，不一定为activity等* @param sticky        是否为sticky事件* @param priority      优先级*/private synchronized void register(Object subscriber, boolean sticky, int priority) {// findSubscriberMethods 查找出该订阅者的所有的onEvent * 方法List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriber.getClass());//将订阅者、事件监听器与事件三者关联起来for (SubscriberMethod subscriberMethod : subscriberMethods) {subscribe(subscriber, subscriberMethod, sticky, priority);}}

以上代码补充一点：EventBus中的事件分为一般事件和 Sticky 事件，相对于一般事件，Sticky 事件不同之处在于，当事件发布后，再有订阅者开始订阅该类型事件，依然能收到该类型事件最近一个 Sticky 事件。

我们走入以上代码的 findSubscriberMethods 部分

 /*** 通过方法名匹配，得到onEvent *  方法，并设置ThreadMode.** @param subscriberClass* @return*/List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {String key = subscriberClass.getName();//存放着当前订阅者的所有的监听事件List<SubscriberMethod> subscriberMethods;//Map<String, List<SubscriberMethod>> methodCache// 提供缓存synchronized (methodCache) {subscriberMethods = methodCache.get(key);}if (subscriberMethods != null) {return subscriberMethods;}subscriberMethods = new ArrayList<SubscriberMethod>();Class<?> clazz = subscriberClass;HashSet<String> eventTypesFound = new HashSet<String>();StringBuilder methodKeyBuilder = new StringBuilder();while (clazz != null) {String name = clazz.getName();//剔除JDK，SDK提供的方法if (name.startsWith("java.") || name.startsWith("javax.") || name.startsWith("android.")) {// Skip system classes, this just degrades performancebreak;}// Starting with EventBus 2.2 we enforced methods to be public (might change with annotations again)//获取类里的所有方法Method[] methods = clazz.getDeclaredMethods();//遍历所有for (Method method : methods) {String methodName = method.getName();//onEvent前缀匹配if (methodName.startsWith(ON_EVENT_METHOD_NAME)) {int modifiers = method.getModifiers();//获取方法的修饰//判断是否为PUBLIC，MODIFIERS_IGNORE 包含（抽象，静态，桥接,合成,后2者是编译器添加的）if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {// 获取该方法入参Class<?>[] parameterTypes = method.getParameterTypes();if (parameterTypes.length == 1) {// 截取onEvent后面的字符，进入匹配String modifierString = methodName.substring(ON_EVENT_METHOD_NAME.length());ThreadMode threadMode;if (modifierString.length() == 0) {threadMode = ThreadMode.PostThread;} else if (modifierString.equals("MainThread")) {threadMode = ThreadMode.MainThread;} else if (modifierString.equals("BackgroundThread")) {threadMode = ThreadMode.BackgroundThread;} else if (modifierString.equals("Async")) {threadMode = ThreadMode.Async;} else {if (skipMethodVerificationForClasses.containsKey(clazz)) {continue;} else {throw new EventBusException("Illegal onEvent method, check for typos: " + method);}}Class<?> eventType = parameterTypes[0];methodKeyBuilder.setLength(0);methodKeyBuilder.append(methodName);methodKeyBuilder.append('>').append(eventType.getName());//e.g print ： onEventMainThread>OneEventString methodKey = methodKeyBuilder.toString();if (eventTypesFound.add(methodKey)) {// Only add if not already found in a sub classsubscriberMethods.add(new SubscriberMethod(method, threadMode, eventType));}}} else if (!skipMethodVerificationForClasses.containsKey(clazz)) {Log.d(EventBus.TAG, "Skipping method (not public, static or abstract): " + clazz + "."+ methodName);}}}//查询到其父类的所有方法--->otto是不行的clazz = clazz.getSuperclass();}if (subscriberMethods.isEmpty()) {throw new EventBusException("Subscriber " + subscriberClass + " has no public methods called "+ ON_EVENT_METHOD_NAME);} else {synchronized (methodCache) {//加入缓存methodCache.put(key, subscriberMethods);}return subscriberMethods;}}

其实也只是通过我们的字符串匹配而来的。这点上，EventBus应该是参考了“约定优于配置的思想”

找到我们class（订阅者）的所有onEvent *方法（事件监听器）了。继续看我们的 subscribe(subscriber, subscriberMethod, sticky, priority);忘记它做了什么的同学可以继续回滚上去看。

// Must be called in synchronized block/*** * @param subscriber         订阅者* @param subscriberMethod   事件监听器* @param sticky             是否为sticky事件* @param priority           优先级*/private void subscribe(Object subscriber, SubscriberMethod subscriberMethod, boolean sticky, int priority) {//获取当前事件监听器的接受事件类型Class<?> eventType = subscriberMethod.eventType;// subscriptionsByEventType中存放着所有的订阅信息（订阅者和事件监听器）和其对应着的事件CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);Subscription newSubscription = new Subscription(subscriber, subscriberMethod, priority);//关联当前订阅者与事件监听器if (subscriptions == null) {subscriptions = new CopyOnWriteArrayList<Subscription>();subscriptionsByEventType.put(eventType, subscriptions); //关联三者，以event为key} else {if (subscriptions.contains(newSubscription)) {//之前已经关联过就跳出//注意这里可是会报错！可是没有人try catch//以下会提到几个避坑的方法throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "+ eventType);}}// Starting with EventBus 2.2 we enforced methods to be public (might change with annotations again)// subscriberMethod.method.setAccessible(true);int size = subscriptions.size();for (int i = 0; i <= size; i++) {if (i == size || newSubscription.priority > subscriptions.get(i).priority) {//优先级插入subscriptions.add(i, newSubscription);break;}}// get all event that registered in this subscriberList<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);//关联订阅者与事件if (subscribedEvents == null) {subscribedEvents = new ArrayList<Class<?>>();typesBySubscriber.put(subscriber, subscribedEvents);}subscribedEvents.add(eventType);//判断是否为sticky事件if (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);}}}

最后的方法如下

//以下方法如果sticky事件不为空的话，就直接postprivate 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, Looper.getMainLooper() == Looper.myLooper());}}

register的流程就结束了，那么在回顾一下，加深下印象

That is all，第一步就这么完成了。是不是很容易？个人觉得EventBus还是很容易的。

这里顺带给大家讲诉下上文说的报错的点的优化。
这里主要针对项目耦合度太大，继承层级过深的情况

EventBus.getDefault().register(this);`

一般大家都会常使用这样来注册，传入是当前的上下文。可是当你继承父类（该父类已经注册过）的时候，子类再注册，会执行以上的错误，”Subscriber XXXX already registered to event XXXX;
原因是：子类的context与父类是相同的，不允许重复注册，避免方法就是父类写了之后，子类不要写。或通过以下方法判断下

if(!EventBus.getDefault().isRegister(this){ //本质是判断有没关联过当前订阅者EventBus.getDefault().register(this);}`

这里也推荐大家一个写法
在当前类中再定义一个内部类

private class MyBean{//单独的class 可以使得匹配也快点。public void onEventMainThead(OneEvent oneEvent){//do something...}
}

注册时如下定义

MyBean  myBean =new MyBean //全局变量...
EventBus.getDefault().register(myBean);//注册...
EventBus.getDefault().unregister(myBean);//反注册

second step：unregister()

你以为我会说post？No! No! No 简单先来，看看最后的unregister吧，短短几行罢了
代码如下：

 /*** Unregisters the given subscriber from all event classes.*/public synchronized void unregister(Object subscriber) {List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);if (subscribedTypes != null) {for (Class<?> eventType : subscribedTypes) {//接触事件和当前订阅者与事件监听器之间的联系unubscribeByEventType(subscriber, eventType);}//去掉当前监听者typesBySubscriber.remove(subscriber);} else {Log.w(TAG, "Subscriber to unregister was not registered before: " + subscriber.getClass());}}

其中跳到了奇怪的地方，我们再跟进去看看

private void unubscribeByEventType(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--;}}}}

嗯哼~上文注册时关联在subscriptionsByEventType（订阅信息与事件的关联）和typesBySubscriber（事件与订阅者的关联）中的对象在反注册的时候都被remove了。so easy！~

值得一提的事，注册时的异常问题在反注册时候是没有的

third step： post

看到这里同学们也都累了，所以我们先看看post的流程图

假如你觉得ok，我们继续~

了解EventBus类中的PostingThreadState类

  /*** For ThreadLocal, much faster to set (and get multiple values).*/final static class PostingThreadState {final List<Object> eventQueue = new ArrayList<Object>();boolean isPosting;boolean isMainThread;Subscription subscription;Object event;boolean canceled;}

读名知意，我就不多说了。
接下来正式看看我们的post方法

/*** Posts the given event to the event bus.*/public void post(Object event) {// currentPostingThreadState为TheadLocal,获取当前线程的postingStatePostingThreadState postingState = currentPostingThreadState.get();List<Object> eventQueue = postingState.eventQueue;eventQueue.add(event);//将当前事件添加入post队列if (!postingState.isPosting) {//事件是否在分发postingState.isMainThread = Looper.getMainLooper() == Looper.myLooper();//是否在主线程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;}}}

主要post还是在postSingleEvent（）里，我们再尾行，或者尾随看看。
代码如下：

 private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {Class<?> eventClass = event.getClass();boolean subscriptionFound = false;if (eventInheritance) {//事件是否允许继承 默认是true//lookupAllEventTypes()得到当前事件的所有的父类，祖宗类.....与接口,不做深入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);//按位或,为true就一直是true哦}} else {subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);}if (!subscriptionFound) {//如果没有成功post过if (logNoSubscriberMessages) {Log.d(TAG, "No subscribers registered for event " + eventClass);}if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&eventClass != SubscriberExceptionEvent.class) {post(new NoSubscriberEvent(this, event));  //可以统一进行处理}}}

postSingleEventForEventType我们集体的post就放在这个方法里，离大结局不远了。行百里者，半九十！
Go On！

 /*** * @param event            要post的事件* @param postingState* @param eventClass       要post的事件或是其接口或父类* @return*/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 = false;try {postToSubscription(subscription, event, postingState.isMainThread);aborted = postingState.canceled;  //只有调用cancelEventDelivery(Object event)方法才会修改这个值。不做深究} finally {//释放引用操作postingState.event = null;          postingState.subscription = null;postingState.canceled = false;}if (aborted) {break;}}return true;}return false;}

终于要看到关于我们想要看到的post真正要做的事啦，前面都只是判断而已，现在才到处理，看看 postToSubscription这个方法，还记得我们的ThreadMode么？

/**** @param subscription     订阅信息* @param event            事件* @param isMainThread     是否为主线程*/private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {switch (subscription.subscriberMethod.threadMode) {case PostThread://java放射，注意这个可是会直接在当前UI线程中调用的，也就是说执行pos的时候，是会直接阻塞住当前线程的invokeSubscriber(subscription, event);break;case MainThread:if (isMainThread) {invokeSubscriber(subscription, event);} else {mainThreadPoster.enqueue(subscription, event);}break;case BackgroundThread: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);}}

ok，终于处理到我们的事件了，EventBus是如何实现各种ThreadMode的调用呢？这是一个很有意思的点。是我看eventBus的目的之一。

其实所有的原理都只是反射。但是如何实现的呢？先概括如下

在UI线程中时，继承的Handler类，在handleMessage中调用的反射；
在子线程时候，继承Runable类，用Executors.newCachedThreadPool().execute(this)来执行任务。其中run方法里也用了反射。

所以我们先来看看它的反射方法，所有的反射都是用得它：

//此方法在EventBus中void invokeSubscriber(PendingPost pendingPost) {Object event = pendingPost.event;Subscription subscription = pendingPost.subscription;PendingPost.releasePendingPost(pendingPost);if (subscription.active) {invokeSubscriber(subscription, event);}}

事件监听器如何执行的我们明白了，接下来就先介绍下几个Poster类的相同属性的变量。

    private final PendingPostQueue queue; //自己实作的pendingPoster队列，post也是按照顺序来的啦~private final EventBus eventBus;  //只是为了回调

个人觉得这三种poster没有太大的点，很容易看明白，所以
以下贴出我的注释。关于这三种Poster

final class HandlerPoster extends Handler {private final PendingPostQueue queue;private final int maxMillisInsideHandleMessage; //最大处理时间，默认为10msprivate final EventBus eventBus;private boolean handlerActive; //为 true 的时候说明开始处理post队列HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {super(looper);this.eventBus = eventBus;this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;queue = new PendingPostQueue();}void enqueue(Subscription subscription, Object event) {//当PendingPost中pendingPostPool长度>0时，取出第一个，为0就直接new一个pendingPost对象返回PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);synchronized (this) {//将pendingPost加入PendingPostQueuequeue.enqueue(pendingPost);if (!handlerActive) {handlerActive = true;if (!sendMessage(obtainMessage())) {throw new EventBusException("Could not send handler message");}}}}@Overridepublic void handleMessage(Message msg) {boolean rescheduled = false;try {long started = SystemClock.uptimeMillis();while (true) {PendingPost pendingPost = queue.poll();if (pendingPost == null) {synchronized (this) {// Check again, this time in synchronizedpendingPost = queue.poll();if (pendingPost == null) {handlerActive = false;return;}}}//还是执行了反射，eventBus留在这的意义....好单薄啊....eventBus.invokeSubscriber(pendingPost);long timeInMethod = SystemClock.uptimeMillis() - started;if (timeInMethod >= maxMillisInsideHandleMessage) { //大于10ms也再继续执行一次if (!sendMessage(obtainMessage())) {//再继续执行throw new EventBusException("Could not send handler message");}//重新安排rescheduled = true;return;}}} finally {handlerActive = rescheduled;}}
}

final class BackgroundPoster implements Runnable {private final PendingPostQueue queue;private final EventBus eventBus;private volatile boolean executorRunning;  //线程池是否在运行BackgroundPoster(EventBus eventBus) {this.eventBus = eventBus;queue = new PendingPostQueue();}public void enqueue(Subscription subscription, Object event) {PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);synchronized (this) {queue.enqueue(pendingPost);if (!executorRunning) {executorRunning = true;//调用eventBus里的executorService。eventBus.getExecutorService().execute(this);}}}@Overridepublic void run() {try {try {while (true) {//队列进栈的时候会notifyAll,此处队列会wait 1m。PendingPost pendingPost = queue.poll(1000);if (pendingPost == null) {synchronized (this) {// Check again, this time in synchronizedpendingPost = queue.poll();if (pendingPost == null) {//不在执行中executorRunning = false;return;}}}eventBus.invokeSubscriber(pendingPost);}} catch (InterruptedException e) {Log.w("Event", Thread.currentThread().getName() + " was interruppted", e);}} finally {//不在执行中executorRunning = false;}}}

3 看过上面的，这下面这个是最简单的，没什么营养

class AsyncPoster implements Runnable {private final PendingPostQueue queue;private final EventBus eventBus;AsyncPoster(EventBus eventBus) {this.eventBus = eventBus;queue = new PendingPostQueue();}public void enqueue(Subscription subscription, Object event) {PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);queue.enqueue(pendingPost);eventBus.getExecutorService().execute(this);}@Overridepublic void run() {PendingPost pendingPost = queue.poll();if(pendingPost == null) {throw new IllegalStateException("No pending post available");}eventBus.invokeSubscriber(pendingPost);}}

这是我第一份讲解源码，如果有解释的不明白，或者模棱两可的地方，希望各位看官记得指正我。我还很弱，要专心练剑。

附上以上内容的源码链接

That is all~~ Thanks！

总结：作为时间总线的设计，看着一遍下来，应该也能知道个大概，特别是继承属性与Sticky事件，平时几乎用不到，不看源码，一般都不知道。精华的点是在如何处理ThreadMode吧。这种思路，对我这种菜鸡而言，还是第一次见，很受启发。

后感：EventBus的源码十分简单，很适合作为大家看的第一份源码。源码是读的过程中是枯燥的。不过收获的丰富的。