EventBus 源码解析

EventBus 源码解析

这篇文章在默认都会使用eventbus的基础上。

一句话概括 利用map保存activty注册的Event方法，通过反射在需要时候运行这些Event，不同的线程运行利用handler 和 runable 转换。

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
mContext = this;
EventBus.getDefault().register(this);

EventBus.getDefault().post(new Item("hello EventBus"));
}

public void onEventMainThread(Item event)
{
Toast.makeText(this,event.content,Toast.LENGTH_SHORT).show();
}

@Override
public void onDestroy()
{
super.onDestroy();
// Unregister
EventBus.getDefault().unregister(this);
}

EventBus.getDefault()其实就是个单例，和我们传统的getInstance一个意思：

public static EventBus getDefault() {
if (defaultInstance == null) {
synchronized (EventBus.class) {
if (defaultInstance == null) {
defaultInstance = new EventBus();
}
}
}
return defaultInstance;
}

注册：register

EventBus.getDefault().register(this);

这句代码的内部就是把 当前activity实例 还有利用反射获取的onEvent开头的方法保存到map中。后来给 EventBus.getDefault().post 反射调用用。

现在看下register的代码

public void register(Object subscriber) {
register(subscriber, false, 0);
}

内部调用了 register(subscriber, false, 0) 方法，其实其他几个注册方法 也是调用了这个方法。

public void register(Object subscriber) {
register(subscriber, false, 0);
}

public void register(Object subscriber, int priority) {
register(subscriber, false, priority);
}

public void registerSticky(Object subscriber) {
register(subscriber, true, 0);
}
public void registerSticky(Object subscriber, int priority) {
register(subscriber, true, priority);
}

看下 register(subscriber, false, 0)的具体内部代码

private synchronized void register(Object subscriber, boolean sticky, int priority) {
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriber.getClass());
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod, sticky, priority);
}
}

三个参数

subscriber 是我们扫描类的对象，也就是我们代码中常见的this;

sticky 这个参数，待会解释

priority 优先级，优先级越高，在调用的时候会越先调用。

看下列代码

List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriber.getClass());

就是方法 findSubscriberMethods 利用java的反射获取 acticity及其父类中所有onEvent 开头的所有方法、还有运行线程（利用onEvent后面的字符串，这里是 MainThread 主线程信息）、activity信息 的list列表

看SubscriberMethod类，就是一个bean

final class SubscriberMethod {
final Method method;
final ThreadMode threadMode;
final Class<?> eventType;
/** Used for efficient comparison */
String methodString;

看下findSubscriberMethods方法

List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {

//开始初始化一些缓存区信息

String key = subscriberClass.getName();
List<SubscriberMethod> subscriberMethods;
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();
if (name.startsWith("java.") || name.startsWith("javax.") || name.startsWith("android.")) {
// Skip system classes, this just degrades performance
break;
}

//去得到所有的方法：
Method[] methods = clazz.getDeclaredMethods();
for (Method method : methods) {  //遍历所有activity中的方法
String methodName = method.getName();
if (methodName.startsWith(ON_EVENT_METHOD_NAME)) {//这边的ON_EVENT_METHOD_NAME 就是字符串onEvent 过滤出所有onEvent开头的方法
int modifiers = method.getModifiers();
if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
Class<?>[] parameterTypes = method.getParameterTypes();//获取参数类型 后来提供post的时候 利用post的参数找到现在注册的onEvent方法
if (parameterTypes.length == 1) {
String modifierString = methodName.substring(ON_EVENT_METHOD_NAME.length());
ThreadMode threadMode;
//下面是用来 这个onEvent是什么结尾的字符串 区分在什么线程中执行
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());
String methodKey = methodKeyBuilder.toString();
if (eventTypesFound.add(methodKey)) {
// 用SubscriberMethod 包装onEvent threadMode执行线程，参数类型 然后加入到list中  返回

subscriberMethods.add(new SubscriberMethod(method, threadMode, eventType));
}
}
} else if (!skipMethodVerificationForClasses.containsKey(clazz)) {
Log.d(EventBus.TAG, "Skipping method (not public, static or abstract): " + clazz + "."
+ methodName);
}
}
}
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;
}
}

继续 遍历onEvent组成的信息的list ,调用subscribe

for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod, sticky, priority);
}

我们看下subscribe内部

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod, boolean sticky, int priority) {
Class<?> eventType = subscriberMethod.eventType;
//初始化
CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
Subscription newSubscription = new Subscription(subscriber, subscriberMethod, priority);
if (subscriptions == null) {
subscriptions = new CopyOnWriteArrayList<Subscription>();
//利用onEvent参数未key 保存这个onEvent的有关信息（前面说过），为后来post的时候
//利用post的参数来查找这个onEvent
subscriptionsByEventType.put(eventType, subscriptions);
} else {
if (subscriptions.contains(newSubscription)) {
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();
//这里是注册的时候提供了优先级，排列到list考前的位置，查找的时候会被先找到
for (int i = 0; i <= size; i++) {
if (i == size || newSubscription.priority > subscriptions.get(i).priority) {
subscriptions.add(i, newSubscription);
break;
}
}

//这里是利用activity为key值保存 所有onEvent的参数list 为不同activty中相同的onEvent的参数 做区分
List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
if (subscribedEvents == null) {
subscribedEvents = new ArrayList<Class<?>>();
typesBySubscriber.put(subscriber, subscribedEvents);
}
subscribedEvents.add(eventType);
//判断sticky；如果为true，从stickyEvents中根据eventType去查找有没有stickyEvent，如果有则立即发布去执行。stickyEvent其实就是我们post时的参数
if (sticky) {
Object stickyEvent;
synchronized (stickyEvents) {
stickyEvent = stickyEvents.get(eventType);
}
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());
}
}
}

post 发送信息

现在很容易就能想到register时，把方法存在subscriptionsByEventType；那么post肯定会去subscriptionsByEventType去取方法，然后调用。

public void post(Object event) {
//currentPostingThreadState是一个ThreadLocal类型的，里面存储了PostingThreadState；PostingThreadState包含了一个eventQueue和一些标志位。
PostingThreadState postingState = currentPostingThreadState.get();
List<Object> eventQueue = postingState.eventQueue;
eventQueue.add(event);

if (!postingState.isPosting) {
//判断当前线程是主线程（UI） 还是其他线程
postingState.isMainThread = Looper.getMainLooper() == Looper.myLooper();
postingState.isPosting = true;
if (postingState.canceled) {
throw new EventBusException("Internal error. Abort state was not reset");
}
try {
//遍历队列中的所有的event，调用postSingleEvent（eventQueue.remove(0), postingState）方法
while (!eventQueue.isEmpty()) {
postSingleEvent(eventQueue.remove(0), postingState);
}
} finally {
postingState.isPosting = false;
postingState.isMainThread = false;
}
}
}

private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {
@Override
protected PostingThreadState initialValue() {
return new PostingThreadState();
}
}

把我们传入的需要post参数，保存到了当前线程中的一个变量PostingThreadState的eventQueue中。就有事一个list。

下面看postSingleEvent

private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
if (eventInheritance) {
//根据传入参数event的Class，去得到一个List<Class<?>>；其实就是得到event当前对象的Class 前面register保存过的
List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
int countTypes = eventTypes.size();
//遍历所有的Class，到subscriptionsByEventType去查找subscriptions 就是 我们前面保存的onEvent 方法的信息 在这边执行
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) {
Log.d(TAG, "No subscribers registered for event " + eventClass);
}
if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
eventClass != SubscriberExceptionEvent.class) {
post(new NoSubscriberEvent(this, event));
}
}
}

下面我们来看看 postSingleEventForEventType 方法,就是遍历每个activity 中onEvent的信息subscriptions 来反射执行

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;
} finally {
postingState.event = null;
postingState.subscription = null;
postingState.canceled = false;
}
if (aborted) {
break;
}
}
return true;
}
return false;
}

终于 到了执行的地方 postToSubscription,看下 我们本例子中执行的onEvent 就是 MainThread，因为现在的线程是主线程 ，onEventMainThread 也是在主线程中执行，所以直接调用invokeSubscriber 反射执行onEvent方法 就行了。

private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
switch (subscription.subscriberMethod.threadMode) {
case PostThread:
invokeSubscriber(subscription, event);
break;
case MainThread:
if (isMainThread) {
invokeSubscriber(subscription, event);
} else {
//当前线程如果不在主线程，就要调用hander了。mainThreadPoster就是一个hander
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);
}
}

———————-以下摘抄网上的 懒得写了—————

case MainThread:

首先去判断当前如果是UI线程，则直接调用；否则： mainThreadPoster.enqueue(subscription, event);把当前的方法加入到队列，然后直接通过handler去发送一个消息，在handler的handleMessage中，去执行我们的方法。说白了就是通过Handler去发送消息，然后执行的。

case BackgroundThread:

如果当前非UI线程，则直接调用；如果是UI线程，则将任务加入到后台的一个队列，最终由Eventbus中的一个线程池去调用

executorService = Executors.newCachedThreadPool();。

case Async:将任务加入到后台的一个队列，最终由Eventbus中的一个线程池去调用；线程池与BackgroundThread用的是同一个。

这么说BackgroundThread和Async有什么区别呢？

BackgroundThread中的任务，一个接着一个去调用，中间使用了一个布尔型变量handlerActive进行的控制。

Async则会动态控制并发。