背景
想着公司项目大量使用EventBus来通信，那么显然有必要熟悉其代码构成，这里边学习边记录。
官方文档
学一门开源技术，官方文档是必不可少的，看完后，再做自己的总结，吸收成自己的东西。
使用方式
这里直接贴官方的用例，非常简单。
- 定义事件Event，也就是将来需要被监听的。
- 实现订阅者的方法，添加好对应的@Annotation，这是将来会被回调到的地方，同时开始订阅register。
- 发送事件。
源码细读
EventBus的基本结构
EventBus是一个总线结构，其所有的操作，都在EventBus这个对象里完成，通常我们都是用的
EventBus.getDefault() 这个对象。
官方结构图
注册：
先从注册开始，EventBus.getDefault().register(this);
/**
* Registers the given subscriber to receive events. Subscribers must call {@link #unregister(Object)} once they
* are no longer interested in receiving events.
* <p/>
* Subscribers have event handling methods that must be annotated by {@link Subscribe}.
* The {@link Subscribe} annotation also allows configuration like {@link
* ThreadMode} and priority.
*/
public void register(Object subscriber) {
//获取当前订阅者的class。
Class<?> subscriberClass = subscriber.getClass();
//查找订阅者那些被@Subscribe标记的方法，用于后续的事件通知。
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
//对单个需要被注册的方法进行注册。
subscribe(subscriber, subscriberMethod);
}
}
}
接下来，进入查找List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
SubscriberMethodFinder 封装了一系列生成订阅者SubscriberMethod的方法，接着上文的思路，进入查询方法：
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
//从缓存中读取，下次register的时候能快速读取。
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}
//是否按反射查找，还是按新增的Index方式查找，默认是false。
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;
}
}
findUsingInfo两个功能类似findUsingReflection，但前者除了会从Index方式找，而且会从反射的方式找，所以这里主要介绍findUsingInfo。
private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
//对查询SubscriberMethod所需方法进行封装的FindState。
//这里还用到里对象池，回收使用FindState对象。
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
//循环查找，会查询subscriberClass父类里头被@Subscribe标记的方法。
while (findState.clazz != null) {
//先通过Index来找。
findState.subscriberInfo = getSubscriberInfo(findState);
if (findState.subscriberInfo != null) {
SubscriberMethod[] array = findState.subscriberInfo.getSubscriberMethods();
for (SubscriberMethod subscriberMethod : array) {
//去掉重复的，方法method名字和事件Event名字公共组成key，查询是否有重复，且只留父类方法。
if (findState.checkAdd(subscriberMethod.method, subscriberMethod.eventType)) {
findState.subscriberMethods.add(subscriberMethod);
}
}
} else {
//如果Index找不到，则通过反射来找。
findUsingReflectionInSingleClass(findState);
}
//findState里进入clazz的父类。
findState.moveToSuperclass();
}
//返回查询到的全部方法。
return getMethodsAndRelease(findState);
}
进入单个注册subscribe(subscriber, subscriberMethod)方法:
// Must be called in synchronized block
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
//找到注册事件的class。
Class<?> eventType = subscriberMethod.eventType;
//用一个Subscription包装注册对象和注册方法。
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
//查询缓存，这里可以看出，如果一个activity注册了event bus，如果不调用解注册，那就可能内存泄露。
CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions == null) {
subscriptions = new CopyOnWriteArrayList<>();
subscriptionsByEventType.put(eventType, subscriptions);
} else {
//避免重复注册，Subscription还重写了equals方法，让比较更准。
if (subscriptions.contains(newSubscription)) {
throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
+ eventType);
}
}
// 根据注册的优先级，在一个event的注册队列里添加注册信息，还能根据优先级插队
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) {
// 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);
}
}
}
这段就是缓存事件，接下来就要进入发送事件的源码。
发送：
/** Posts the given event to the event bus. */
public void post(Object event) {
//读取当前线程的PostingThreadState，因为currentPostingThreadState是一个TreadLocal，这是一个存储线程单独数据的变量。
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;
}
}
}
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
//eventInheritance默认是true
if (eventInheritance) {
//查找父类的event
//其实就是在发送一个event时，也会通知到监听其父类event的subscriber
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) {
//如果事件没有被注册，则会变成NoSubscriberEvent发送出去
post(new NoSubscriberEvent(this, event));
}
}
}
//正式开始通知订阅者
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;
}
//这里会进行线程切换操作
private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
switch (subscription.subscriberMethod.threadMode) {
case POSTING:
//直接和通知者线程一致
invokeSubscriber(subscription, event);
break;
case MAIN:
//切换到主线程，用到了Handler
if (isMainThread) {
invokeSubscriber(subscription, event);
} else {
mainThreadPoster.enqueue(subscription, event);
}
break;
case MAIN_ORDERED:
//切换到主线程，用到了Handler，但是优先用队列
if (mainThreadPoster != null) {
mainThreadPoster.enqueue(subscription, event);
} else {
// temporary: technically not correct as poster not decoupled from subscriber
invokeSubscriber(subscription, event);
}
break;
case BACKGROUND:
//切换到Background线程
if (isMainThread) {
backgroundPoster.enqueue(subscription, event);
} else {
invokeSubscriber(subscription, event);
}
break;
case ASYNC:
//放入event bus的线程池执行
asyncPoster.enqueue(subscription, event);
break;
default:
throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
}
}
//真正通知订阅者的地方，利用反射通知
void invokeSubscriber(Subscription subscription, Object event) {
try {
subscription.subscriberMethod.method.invoke(subscription.subscriber, event);
} catch (InvocationTargetException e) {
//InvocationTargetException可以catch住反射方法本身产生的异常，所以需要特别注意，
//有时候可以功能不正确，但不清楚问题在哪，很有可能是方法本身产生了异常，但被event bus捕获了。
handleSubscriberException(subscription, event, e.getCause());
} catch (IllegalAccessException e) {
throw new IllegalStateException("Unexpected exception", e);
}
}
这里多加一段，就是EventBus对异常任务的处理：
//EventBus提供的异常处理方式，有logger，throwSubscriberException抛出异常，sendSubscriberExceptionEvent订阅异常
//建议实现一个对SubscriberExceptionEvent的订阅，这个能很方便的定位EventBus执行产生的异常
private void handleSubscriberException(Subscription subscription, Object event, Throwable cause) {
if (event instanceof SubscriberExceptionEvent) {
if (logSubscriberExceptions) {
// Don't send another SubscriberExceptionEvent to avoid infinite event recursion, just log
logger.log(Level.SEVERE, "SubscriberExceptionEvent subscriber " + subscription.subscriber.getClass()
+ " threw an exception", cause);
SubscriberExceptionEvent exEvent = (SubscriberExceptionEvent) event;
logger.log(Level.SEVERE, "Initial event " + exEvent.causingEvent + " caused exception in "
+ exEvent.causingSubscriber, exEvent.throwable);
}
} else {
if (throwSubscriberException) {
throw new EventBusException("Invoking subscriber failed", cause);
}
if (logSubscriberExceptions) {
logger.log(Level.SEVERE, "Could not dispatch event: " + event.getClass() + " to subscribing class "
+ subscription.subscriber.getClass(), cause);
}
if (sendSubscriberExceptionEvent) {
SubscriberExceptionEvent exEvent = new SubscriberExceptionEvent(this, cause, event,
subscription.subscriber);
post(exEvent);
}
}
}
最后就是解注册了。
解除注册
//取消订阅
/** Unregisters the given subscriber from all event classes. */
public synchronized void unregister(Object subscriber) {
//找到这个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());
}
}
一些用法介绍
除开基本的注册、发送事件、解除注册，EventBus还提供了很多别的用法
SubscriberExceptionEvent使用
从前面的handleSubscriberException可以看到，如果用户去订阅的方法有异常，是没法直接看到的，所以当事件注册非常多时，有时候难以定位问题，这里就有个小方法：
@Subscribe(threadMode = ThreadMode.BACKGROUND)
public void onEventException(SubscriberExceptionEvent exceptionEvent){
//仅Debug版本下有效
if (GlobalConfig.DEBUG){
StringBuffer buffer = new StringBuffer();
buffer.append(" \n");
//描述异常事件
buffer.append("Event: ").append(exceptionEvent.causingEvent.getClass()).append("\n");
buffer.append("Subscriber: ").append(exceptionEvent.causingSubscriber.getClass()).append("\n");
//描述异常信息
buffer.append("Exception Info: ").append(exceptionEvent.throwable.getMessage()).append("\n");
//描述异常堆栈
buffer.append("------ begin of trace ------\n");
for (StackTraceElement element : exceptionEvent.throwable.getStackTrace()) {
buffer.append(element.toString()).append("\n");
}
buffer.append("------ end of trace ------\n");
Log.e("EventBus",buffer.toString());
}
}
粘性事件
@Subscribe(threadMode = ThreadMode.BACKGROUND,sticky = true)
public void onEventMainThread(TestFinishedEvent event) {
//do something
}
粘性事件，使用非常简单，就是再标记方法的@Subscribe设置sticky为ture。
在有些场景下，由于时间问题，事件已经发布了，但是注册还未开始，如果依然依赖早期事件，那么粘性事件就是非常有用的了。从源码上来看，在注册事件时，会触发一下之前保持的粘性事件。
