eventbus异步框架源码解析

一、EventBus使用

1、首先是注册

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

2、响应事件方法

1  @Subscribe(threadMode = ThreadMode.BACKGROUND, sticky = true, priority = 100)
2     public void jiaoTest(String str) {
3         System.out.println("响应方法：" + str);
4     }

参数解析：

threadMode ：方法执行的线程

sticky:是否接受粘性事件

priority:优先级

String str:方法接受对象类型

3、事件分发

1  EventBus.getDefault().post("Test");

4、解除注册

1 EventBus.getDefault().unregister(this);

以上就是EventBus的使用过程，用起来非常简单方便，非常实用。

二、注册源码解析

对应以上的注册方式，我们就从EventBus.getDefault().register(this);入手，首先查看EventBus.getDefault()

看看EventBus是如何初始化的；

 1 /** Convenience singleton for apps using a process-wide EventBus instance. */
 2     public static EventBus getDefault() {
 3         if (defaultInstance == null) {
 4             synchronized (EventBus.class) {
 5                 if (defaultInstance == null) {
 6                     defaultInstance = new EventBus();
 7                 }
 8             }
 9         }
10         return defaultInstance;
11     }

可以看出来，EventBus是单例模式存在的，一个项目中只能有一个EventBus这样有利于管理订阅者和订阅方法，这会在下面的介绍中体现出来。

接下来看register(this)

 1 public void register(Object subscriber) {
 2         //订阅者
 3         Class<?> subscriberClass = subscriber.getClass();
 4         List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
 5         synchronized (this) {
 6             for (SubscriberMethod subscriberMethod : subscriberMethods) {
 7                 subscribe(subscriber, subscriberMethod);
 8             }
 9         }
10     }
11

可以看出首先获取订阅者的类对象Class<?> subscriberClass = subscriber.getClass();

在看这段代码之前，我们首先要了解SubscriberMethod和subscriberMethodFinder.findSubscriberMethods方法到底做了什么

首先来看SubscriberMethod

 1 public class SubscriberMethod {
 2     final Method method;//方法
 3     final ThreadMode threadMode;//执行线程
 4     final Class<?> eventType;//接收的事件类型
 5     final int priority;//优先级
 6     final boolean sticky;
 7     /** Used for efficient comparison */
 8     String methodString;
 9 ....
10 }

可以看出SubscriberMethod其实就是一个订阅方法的实体类，里面保存了订阅方法信息

接着看subscriberMethodFinder.findSubscriberMethods

该方法的作用其实就是从订阅类中获取所有的订阅方法信息；

 1 List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
 2      
 3         //首先从缓存中读取 
 4         List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
 5         if (subscriberMethods != null) {
 6             return subscriberMethods;
 7         }
 8         
 9 
10         //是否忽略注解器生成的MyEventBusIndex类
11 
12         if (ignoreGeneratedIndex) {
13             //利用反射来获取订阅类中的订阅方法信息
14             subscriberMethods = findUsingReflection(subscriberClass);
15         } else {
16             //从注解器生成的MyEventBusIndex类中获得订阅类的订阅方法信息
17             subscriberMethods = findUsingInfo(subscriberClass);
18         }
19         if (subscriberMethods.isEmpty()) {
20             throw new EventBusException("Subscriber " + subscriberClass
21                     + " and its super classes have no public methods with the @Subscribe annotation");
22         } else {
23             //保存进缓存
24             METHOD_CACHE.put(subscriberClass, subscriberMethods);
25             return subscriberMethods;
26         }
27     }

我们看到，该方法首先从缓存中获取订阅类的订阅方法信息，如果没有则通过两种方式来获取

1、通过EventBusAnnotationProcessor（注解处理器）生成的MyEventBusIndex中获取
2、利用反射来读取订阅类中订阅方法信息

EventBusAnnotationProcessor是什么东东？（此处参考：文／达达达达sky（简书作者）原文链接：http://www.jianshu.com/p/f057c460c77e）

在3.0版本中,EventBus提供了一个EventBusAnnotationProcessor注解处理器来在编译期通过读取@Subscribe()注解并解析,
处理其中所包含的信息,然后生成java类来保存所有订阅者关于订阅的信息,这样就比在运行时使用反射来获得这些订阅者的
信息速度要快.我们可以参考EventBus项目里的EventBusPerformance这个例子,编译后我们可以在build文件夹里找到这个类
,MyEventBusIndex 类,当然类名是可以自定义的.我们大致看一下生成的MyEventBusIndex类是什么样的:

 1 /**
 2  * This class is generated by EventBus, do not edit.
 3  */
 4 public class MyEventBusIndex implements SubscriberInfoIndex {
 5     private static final Map<Class<?>, SubscriberInfo> SUBSCRIBER_INDEX;
 6 
 7     static {
 8         SUBSCRIBER_INDEX = new HashMap<Class<?>, SubscriberInfo>();
 9 
10         putIndex(new SimpleSubscriberInfo(org.greenrobot.eventbusperf.testsubject.PerfTestEventBus.SubscriberClassEventBusAsync.class,
11                 true, new SubscriberMethodInfo[]{
12                 new SubscriberMethodInfo("onEventAsync", TestEvent.class, ThreadMode.ASYNC),
13         }));
14 
15         putIndex(new SimpleSubscriberInfo(TestRunnerActivity.class, true, new SubscriberMethodInfo[]{
16                 new SubscriberMethodInfo("onEventMainThread", TestFinishedEvent.class, ThreadMode.MAIN),
17         }));
18     }
19 
20     private static void putIndex(SubscriberInfo info) {
21         SUBSCRIBER_INDEX.put(info.getSubscriberClass(), info);
22     }
23 
24     @Override
25     public SubscriberInfo getSubscriberInfo(Class<?> subscriberClass) {
26         SubscriberInfo info = SUBSCRIBER_INDEX.get(subscriberClass);
27         if (info != null) {
28             return info;
29         } else {
30             return null;
31         }
32     }
33 }

可以看出是使用一个静态HashMap即:SUBSCRIBER_INDEX来保存订阅类的信息,其中包括了订阅类的class对象,

是否需要检查父类,以及订阅方法的信息SubscriberMethodInfo的数组,SubscriberMethodInfo中又保存了,订阅方法的方法名,

订阅的事件类型,触发线程,是否接收sticky事件以及优先级priority.这其中就保存了register()的所有需要的信息；

我们重点研究一下通过反射来获取订阅方法信息即：findUsingReflection(subscriberClass);

1  private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
2         FindState findState = prepareFindState();
3         findState.initForSubscriber(subscriberClass);
4         while (findState.clazz != null) {
5             findUsingReflectionInSingleClass(findState);
6             findState.moveToSuperclass();
7         }
8         return getMethodsAndRelease(findState);
9     }

FindState其实就是一个里面保存了订阅者和订阅方法信息的一个实体类，包括订阅类中所有订阅的事件类型和所有的订阅方法等。

我们看到会首先创建一个FindState对象并执行findUsingReflectionInSingleClass(findState);来获取订阅类的方法信息

 1 private void findUsingReflectionInSingleClass(FindState findState) {
 2         Method[] methods;
 3         try {
 4             // This is faster than getMethods, especially when subscribers are fat classes like Activities
 5             methods = findState.clazz.getDeclaredMethods();
 6         } catch (Throwable th) {
 7             // Workaround for java.lang.NoClassDefFoundError, see https://github.com/greenrobot/EventBus/issues/149
 8             //通过反射获取到订阅类中的所有方法
 9             methods = findState.clazz.getMethods();
10             findState.skipSuperClasses = true;
11         }
12         //遍历所有方法，忽略private类型的，最后如果是公有，并且不是 
13         //java编译器 生成的方法名，那么就是我们要的了。
14         for (Method method : methods) {
15             int modifiers = method.getModifiers();
16             if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
17                 Class<?>[] parameterTypes = method.getParameterTypes();
18                 //保证只有一个事件参数
19                 if (parameterTypes.length == 1) {
20                     //得到注解
21                     Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
22                     if (subscribeAnnotation != null) {
23                         Class<?> eventType = parameterTypes[0];
24                         //校验是否添加该方法
25                         if (findState.checkAdd(method, eventType)) {
26                             ThreadMode threadMode = subscribeAnnotation.threadMode();
27                             //添加方法
28                             findState.subscriberMethods.add(new SubscriberMethod(method, eventType, threadMode,
29                                     subscribeAnnotation.priority(), subscribeAnnotation.sticky()));
30                         }
31                     }
32                 } else if (strictMethodVerification && method.isAnnotationPresent(Subscribe.class)) {
33                     String methodName = method.getDeclaringClass().getName() + "." + method.getName();
34                     throw new EventBusException("@Subscribe method " + methodName +
35                             "must have exactly 1 parameter but has " + parameterTypes.length);
36                 }
37             } else if (strictMethodVerification && method.isAnnotationPresent(Subscribe.class)) {
38                 String methodName = method.getDeclaringClass().getName() + "." + method.getName();
39                 throw new EventBusException(methodName +
40                         " is a illegal @Subscribe method: must be public, non-static, and non-abstract");
41             }
42         }
43     }

可以看到，首先会得到订阅类的class对象并通过反射获取订阅类中的所有方法信息，然后通过筛选获取到订阅方法集合。

程序执行到此我们就获取到了订阅类中的所有的订阅方法信息，接下来我们就要对订阅方法进行注册；

subscribe(subscriber, subscriberMethod);//参数：1订阅者2订阅方法集

 1  // Must be called in synchronized block
 2     private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
 3         //获取订阅方法的参数类型
 4         Class<?> eventType = subscriberMethod.eventType;
 5         Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
 6         //根据订阅的事件类型获取所有的订阅者
 7         CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
 8         //将订阅者添加到subscriptionsByEventType集合中
 9         if (subscriptions == null) {
10             subscriptions = new CopyOnWriteArrayList<>();
11             subscriptionsByEventType.put(eventType, subscriptions);
12         } else {
13             if (subscriptions.contains(newSubscription)) {
14                 throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
15                         + eventType);
16             }
17         }
18 
19         //根据优先级，将订阅者插入到指定的位置
20         int size = subscriptions.size();
21         for (int i = 0; i <= size; i++) {
22             if (i == size || subscriberMethod.priority > subscriptions.get(i).subscriberMethod.priority) {
23                 subscriptions.add(i, newSubscription);
24                 break;
25             }
26         }
27         
28         //获取订阅者所有订阅的事件类型
29 
30         List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
31         if (subscribedEvents == null) {
32             subscribedEvents = new ArrayList<>();
33             typesBySubscriber.put(subscriber, subscribedEvents);
34         }
35         //将该事件类型添加到typesBySubscriber中
36         subscribedEvents.add(eventType);
37 
38 
39         //如果接收sticky事件,立即分发sticky事件
40         if (subscriberMethod.sticky) {
41             if (eventInheritance) {
42                 // Existing sticky events of all subclasses of eventType have to be considered.
43                 // Note: Iterating over all events may be inefficient with lots of sticky events,
44                 // thus data structure should be changed to allow a more efficient lookup
45                 // (e.g. an additional map storing sub classes of super classes: Class -> List<Class>).
46                 Set<Map.Entry<Class<?>, Object>> entries = stickyEvents.entrySet();
47                 for (Map.Entry<Class<?>, Object> entry : entries) {
48                     Class<?> candidateEventType = entry.getKey();
49                     if (eventType.isAssignableFrom(candidateEventType)) {
50                         Object stickyEvent = entry.getValue();
51                         checkPostStickyEventToSubscription(newSubscription, stickyEvent);
52                     }
53                 }
54             } else {
55                 Object stickyEvent = stickyEvents.get(eventType);
56                 checkPostStickyEventToSubscription(newSubscription, stickyEvent);
57             }
58         }
59     }

上面这段代码涉及到几个对象我来介绍一下：

Subscription

//订阅者信息
final class Subscription {
final Object subscriber;//订阅者
final SubscriberMethod subscriberMethod;//订阅方法
}

subscriptionsByEventType
key订阅方法类型 values 所有订阅了该类型的订阅者集合
Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;

typesBySubscriber
key订阅者 values订阅事件集合
Map<Object, List<Class<?>>> typesBySubscriber;

了解了这几个对象，上面的代码就很容易看懂了，

1、首先获取订阅方法的参数类型即订阅事件类型

2、根据订阅事件类型获取该事件类型的所有订阅者

3、将该订阅者添加到该事件类型的订阅者集合中即：subscriptionsByEventType

4、获取订阅者所有的订阅事件类型

5、将该事件类型添加到该订阅者的订阅事件类型集中即:typesBySubscriber

至此，就完成了订阅类中订阅方法的注册，我们来看一下整个流程

eventbus异步框架源码解析

三、事件分发解析

接下来我们来分析EventBus的事件分发机制即：EventBus.getDefault().post("Test");

我们从post方法入手

 1 /** Posts the given event to the event bus. */
 2     public void post(Object event) {
 3         //获取当前线程的postingState
 4         PostingThreadState postingState = currentPostingThreadState.get();
 5         //取得当前线程的事件队列
 6         List<Object> eventQueue = postingState.eventQueue;
 7         //将该事件添加到当前的事件队列中等待分发
 8         eventQueue.add(event);
 9 
10         if (!postingState.isPosting) {
11             //判断是否是在主线程post
12             postingState.isMainThread = Looper.getMainLooper() == Looper.myLooper();
13             postingState.isPosting = true;
14             if (postingState.canceled) {
15                 throw new EventBusException("Internal error. Abort state was not reset");
16             }
17             try {
18                 while (!eventQueue.isEmpty()) {
19                     //分发事件
20                     postSingleEvent(eventQueue.remove(0), postingState);
21                 }
22             } finally {
23                 postingState.isPosting = false;
24                 postingState.isMainThread = false;
25             }
26         }
27     }

什么是PostingThreadState？

1 final static class PostingThreadState {
2         final List<Object> eventQueue = new ArrayList<Object>();//当前线程的事件队列
3         boolean isPosting;//是否有事件正在分发
4         boolean isMainThread;//post的线程是否是主线程
5         Subscription subscription;//订阅者
6         Object event;//订阅事件
7         boolean canceled;//是否取消
8     }

PostingThreadState中包含了当前线程的事件队列，就是当前线程所有分发的事件都保存在eventQueue事件队列中

以及订阅者订阅事件等信息，有了这些信息我们就可以从事件队列中取出事件分发给对应的订阅者。

PostingThreadState怎么获得？

1 ThreadLocal 是一个线程内部的数据存储类，通过它可以在指定的线程中存储数据，而这段数据是不会与其他线程共享的。
2      private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {
3         @Override
4         protected PostingThreadState initialValue() {
5             return new PostingThreadState();
6         }
7     };

可以看出currentPostingThreadState的实现是一个包含了PostingThreadState的ThreadLocal对象,这样可以保证取到的都是

自己线程对应的数据。

我们有了PostingThreadState获取到了当前线程的事件队列，接下来就是事件分发，我们来看

postSingleEvent(eventQueue.remove(0), postingState);

 1 事件分发
 2      private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
 3         //得到事件类型
 4         Class<?> eventClass = event.getClass();
 5         boolean subscriptionFound = false;
 6 
 7         //是否触发订阅了该事件(eventClass)的父类,以及接口的类的响应方法.
 8         if (eventInheritance) {
 9             List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
10             int countTypes = eventTypes.size();
11             for (int h = 0; h < countTypes; h++) {
12                 Class<?> clazz = eventTypes.get(h);
13                 subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
14             }
15         } else {
16             subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
17         }
18         if (!subscriptionFound) {
19             if (logNoSubscriberMessages) {
20                 Log.d(TAG, "No subscribers registered for event " + eventClass);
21             }
22             if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
23                     eventClass != SubscriberExceptionEvent.class) {
24                 post(new NoSubscriberEvent(this, event));
25             }
26         }
27     }

通过以上代码我们可以发现，真正的事件分发是通过postSingleEventForEventType(event, postingState, eventClass);发出去的我们来看：

 1 private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
 2         CopyOnWriteArrayList<Subscription> subscriptions;
 3         synchronized (this) {
 4             //根据事件类型获取所有的订阅者
 5             subscriptions = subscriptionsByEventType.get(eventClass);
 6         }
 7         //向每个订阅者分发事件
 8         if (subscriptions != null && !subscriptions.isEmpty()) {
 9             for (Subscription subscription : subscriptions) {
10                 postingState.event = event;
11                 postingState.subscription = subscription;
12                 boolean aborted = false;
13                 try {
14                     postToSubscription(subscription, event, postingState.isMainThread);
15                     aborted = postingState.canceled;
16                 } finally {
17                     postingState.event = null;
18                     postingState.subscription = null;
19                     postingState.canceled = false;
20                 }
21                 if (aborted) {
22                     break;
23                 }
24             }
25             return true;
26         }
27         return false;
28     }

可以看到首先根据事件类型获取到所有的订阅者，然后循环向每个订阅者发送事件，通过

postToSubscription(subscription, event, postingState.isMainThread);发送出去

 1  private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
 2         switch (subscription.subscriberMethod.threadMode) {
 3             case POSTING://默认的 ThreadMode，表示在执行 Post 操作的线程直接调用订阅者的事件响应方法，
 4             //不论该线程是否为主线程（UI 线程）。
 5                 invokeSubscriber(subscription, event);
 6                 break;
 7             case MAIN://在主线程中执行响应方法。
 8                 if (isMainThread) {
 9                     invokeSubscriber(subscription, event);
10                 } else {
11                     mainThreadPoster.enqueue(subscription, event);
12                 }
13                 break;
14             case BACKGROUND://在后台线程中执行响应方法。
15                 if (isMainThread) {
16                     backgroundPoster.enqueue(subscription, event);
17                 } else {
18                     invokeSubscriber(subscription, event);
19                 }
20                 break;
21             case ASYNC://不论发布线程是否为主线程，都使用一个空闲线程来处理。
22                 asyncPoster.enqueue(subscription, event);
23                 break;
24             default:
25                 throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
26         }
27     }

以上的四种threadMode可以看代码注释简单了解一下，通过一下代码我们来看一下订阅方法最后是通过invokeSubscriber(subscription, event);来执行的

 1 //最终通过反射调用订阅者的订阅函数 并把event作为参数传入
 2      void invokeSubscriber(Subscription subscription, Object event) {
 3         try {
 4             subscription.subscriberMethod.method.invoke(subscription.subscriber, event);
 5         } catch (InvocationTargetException e) {
 6             handleSubscriberException(subscription, event, e.getCause());
 7         } catch (IllegalAccessException e) {
 8             throw new IllegalStateException("Unexpected exception", e);
 9         }
10     }

真相大白；最后是通过反射的方式，调用了订阅类中的订阅方法。我们来总结一下整个事件分发的过程

1、首先获取当前线程的PostingThreadState对象从而获取到当前线程的事件队列

2、通过事件类型获取到所有订阅者集合

3、通过反射执行订阅者中的订阅方法

是不是很简单。

我们来看一下整个事件分发的流程图

eventbus异步框架源码解析

四、取消注册解析

我们简单看一下取消注册的源码EventBus.getDefault().unregister(this);

 1  /** Unregisters the given subscriber from all event classes. */
 2     public synchronized void unregister(Object subscriber) {
 3         //获取订阅者的所有订阅的事件类型
 4         List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
 5         if (subscribedTypes != null) {
 6             for (Class<?> eventType : subscribedTypes) {
 7                 //从事件类型的订阅者集合中移除订阅者
 8                 unsubscribeByEventType(subscriber, eventType);
 9             }
10             typesBySubscriber.remove(subscriber);
11         } else {
12             Log.w(TAG, "Subscriber to unregister was not registered before: " + subscriber.getClass());
13         }
14     }

再来看一下：unsubscribeByEventType(subscriber, eventType);

 1  /** Only updates subscriptionsByEventType, not typesBySubscriber! Caller must update typesBySubscriber. */
 2     private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
 3         //获取事件类型的所有订阅者
 4         List<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
 5         //遍历订阅者集合，将解除的订阅者移除
 6         if (subscriptions != null) {
 7             int size = subscriptions.size();
 8             for (int i = 0; i < size; i++) {
 9                 Subscription subscription = subscriptions.get(i);
10                 if (subscription.subscriber == subscriber) {
11                     subscription.active = false;
12                     subscriptions.remove(i);
13                     i--;
14                     size--;
15                 }
16             }
17         }
18     }
19

总结一下：

1、首先获取订阅者的所有订阅事件

2、遍历订阅事件

3、根据订阅事件获取所有的订阅了该事件的订阅者集合

4、将该订阅者移除

5、将步骤1中的集合中的订阅者移除

这样就完成了取消订阅的全过程；

最后我们从开发者的角度来总结一下EventBus的工作原理

订阅逻辑

1、首先用register（）方法注册一个订阅者

2、获取该订阅者的所有订阅的方法

3、根据该订阅者的所有订阅的事件类型，将订阅者存入到每个以事件类型为key 以所有订阅者为values的map集合中

4、然后将订阅事件添加到以订阅者为key 以订阅者所有订阅事件为values的map集合中

4.1、如果是订阅了粘滞事件的订阅者，从粘滞事件缓存区获取之前发送过的粘滞事件，响应这些粘滞事件。

事件发送逻辑

1、首先获取当前线程的事件队列

2、将要发送的事件添加到事件队列中

3、根据发送事件类型获取所有的订阅者

4、根据响应方法的执行模式，在相应线程通过反射执行订阅者的订阅方法

取消逻辑

1、首先通过unregister方法拿到要取消的订阅者

2、得到该订阅者的所有订阅事件类型

3、遍历事件类型，根据每个事件类型获取到所有的订阅者集合，并从集合中删除该订阅者

4、将订阅者从步骤2的集合中移除

eventbus异步框架源码解析

相关推荐