EventBus3.0 源码解析
2018-05-26 本文已影响1人
导火索
EventBus，是一个基于Publish/Subscribe的事件总线框架。更通俗的讲，EventBus就是一个基于观察者模式的框架，但是EventBus在观察者模式的基础上有很大一部分改进，基本是实现了Publish/Subscribe的解耦。
这里主要对EventBus3.0的实现原理及源码进行分析，关于EventBus的具体用法，可以去GitHub具体查看，这里不做过多介绍。
通过对EventBus的使用，我们知道要想使用EventBus，必须要有三个步骤：
- 定义事件
- 定义事件处理方法并注册（取消注册）
- 发送事件
所以，分析源码的过程也将从这三个步骤中逐一分析。
关于Eventbus3.0的新特性，通过注解处理器创建索引，优化Subscriber的注册，将在下一个篇幅中具体讲解，这里不做过多介绍。
在下面的源码分析中只要涉及到该特性的也暂时忽略。
创建Event及订阅方法
定义Event
定义Event，就是定义一个普通类，这个类将作为消息体通过Publisher发送到Subscriber，这里对于定义事件就不做过多说明了，这个将会在下面的过程中有具体提到。
public static class MessageEvent {
// 如果需要可以定义字段
}
定义订阅方法
要想通过该Event完成具体事件，就必须要声明一个订阅方法，这个方法必须使用注解@Subscribe注释（EventBus只能处理@Subscribe注释的方法），并且该方法有且仅有一个Event的参数，这个在后面的源码分析中会具体提到为什么只能有一个参数。
@Subscribe(threadMode = ThreadMode.MAIN)
public void onMessageEvent(MessageEvent event) {
/* Do something */
};
创建EventBus
订阅方法声明之后，通过EventBus.getDefault().register(this)注册后订阅方法才可以在运行过程中被处理，下面我们看看EventBus.getDefault().register(this)具体完成了那些功能。
getDefault()
EventBus.getDefault()主要是获取一个EventBus实例。
public static EventBus getDefault() {
if (defaultInstance == null) {
synchronized (EventBus.class) {
if (defaultInstance == null) {
defaultInstance = new EventBus();
}
}
}
return defaultInstance;
}
可以看出，这里获取EventBus实例使用一个单例模式并且使用了DCL机制。但是在查看源码时发现EventBusy声明了两个构造方法：一个是公共的无参构造方法、一个是默认类型的带有一个EventBusBuilder参数的构造方法。如下：
/**
* Creates a new EventBus instance; each instance is a separate scope in which events are delivered. To use a
* central bus, consider {@link #getDefault()}.
*/
public EventBus() {
this(DEFAULT_BUILDER);
}
EventBus(EventBusBuilder builder) {
logger = builder.getLogger();
subscriptionsByEventType = new HashMap<>();
typesBySubscriber = new HashMap<>();
stickyEvents = new ConcurrentHashMap<>();
mainThreadSupport = builder.getMainThreadSupport();
mainThreadPoster = mainThreadSupport != null ? mainThreadSupport.createPoster(this) : null;
backgroundPoster = new BackgroundPoster(this);
asyncPoster = new AsyncPoster(this);
indexCount = builder.subscriberInfoIndexes != null ? builder.subscriberInfoIndexes.size() : 0;
subscriberMethodFinder = new SubscriberMethodFinder(builder.subscriberInfoIndexes,
builder.strictMethodVerification, builder.ignoreGeneratedIndex);
logSubscriberExceptions = builder.logSubscriberExceptions;
logNoSubscriberMessages = builder.logNoSubscriberMessages;
sendSubscriberExceptionEvent = builder.sendSubscriberExceptionEvent;
sendNoSubscriberEvent = builder.sendNoSubscriberEvent;
throwSubscriberException = builder.throwSubscriberException;
eventInheritance = builder.eventInheritance;
executorService = builder.executorService;
}
如果我们通过无参构造方法创建一个EventBus实例，这将始终调用构造方法EventBus(EventBusBuilder builder)。
DEFAULT_BUILDER就是一个EventBusBuilder的实例，而这种创建方式和通过getDefault方法创建的EventBus实例没有差别。
默认类型的构造方法具有包级访问权限，对于开发者一般是无法通过该构造方法创建实例的。
通过EventBusBuilder的命名我们就可以看出，这里使用了一个建造者模式，这里我们看看EventBusBuilder的源码。
public class EventBusBuilder {
private final static ExecutorService DEFAULT_EXECUTOR_SERVICE = Executors.newCachedThreadPool();
boolean logSubscriberExceptions = true;
boolean logNoSubscriberMessages = true;
boolean sendSubscriberExceptionEvent = true;
boolean sendNoSubscriberEvent = true;
boolean throwSubscriberException;
boolean eventInheritance = true;
boolean ignoreGeneratedIndex;
boolean strictMethodVerification;
ExecutorService executorService = DEFAULT_EXECUTOR_SERVICE;
List<Class<?>> skipMethodVerificationForClasses;
List<SubscriberInfoIndex> subscriberInfoIndexes;
Logger logger;
MainThreadSupport mainThreadSupport;
// 省略
EventBusBuilder() {
}
/** Builds an EventBus based on the current configuration. */
public EventBus build() {
return new EventBus(this);
}
}
在EventBusBuilder中定义了一系列的属性，这些都是EventBus的一些配置参数，如果想要通过建造者模式创建一个EventBus实例，我们需要通过以下方式：
EventBus eventBus = EventBus.builder()
.logNoSubscriberMessages(false)
.sendNoSubscriberEvent(false)
.build();
一般情况下，如果没有一些特殊的要求，都是不需要通过这种方式创建实例的，在官方文档中都是推荐通过EventBus.getDefault()获取实例。所以对于该种获取EventBus的方法也就不多讲了。
注册Subscriber
在获取EventBus实例后，就可以通过register()注册Subscriber了。
这里我们先说下一个类SubscriberMethod，该类代表了一个订阅方法，在该类中封装了一个订阅方法相关的所有属性。
下面所有的订阅方法都用SubscriberMethod表示以便于理解。
public class SubscriberMethod {
// 方法
final Method method;
// 线程模式
final ThreadMode threadMode;
// 事件类型
final Class<?> eventType;
// 优先级
final int priority;
// 粘性事件标记
final boolean sticky;
}
register()
/**
* 注册subscriber，subscriber必须要调用unregister取消注册<br/>
* subscriber中的事件处理方法必须被 @Subscribe 注解修饰
*/
public void register(Object subscriber) {
// 获取subscriber的具体类型
Class<?> subscriberClass = subscriber.getClass();
// 获取subscriber中所有的事件方法
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}
在register()中，首先获取subscriber的class类型，这就代表了一个Subscriber订阅者。
然后通过subscriberMethodFinder获取该Subscriber中所有的。subscriberMethodFinder是一个SubscriberMethodFinder实例，该类的作用就是用户获取指定Subscriber中所有SubscriberMethod。
findSubscriberMethods
findSubscriberMethods方法的作用就是获取一个Subscriber中所有的SubscriberMethod。
/**
* 获取subscriber中所有的注解方法
*
* @param subscriberClass
* @return
*/
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
// 缓存该subscriber类中的所有事件方法
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}
if (ignoreGeneratedIndex) {
// 这是直接通过反射获取Subscriber
subscriberMethods = findUsingReflection(subscriberClass);
} else {
// 这是添加了Eventbus3.0新特性，通过注解处理器在编译期间通过创建索引优化Subscriber的注册，
subscriberMethods = findUsingInfo(subscriberClass);
}
// 在subscriber类中如果没有注解方法，会抛出异常
if (subscriberMethods.isEmpty()) {
throw new EventBusException("Subscriber中必须定义Event处理方法。 " + subscriberClass
+ " and its super classes have no public methods with the @Subscribe annotation");
} else {
// 存储到Map中
METHOD_CACHE.put(subscriberClass, subscriberMethods);
return subscriberMethods;
}
}
在第一次调用register()时，一般会执行findUsingInfo(subscriberClass)。在findSubscriberMethods中，我们可以看到会抛出一个EventBusException异常，这是一个运行时异常，在Subscriber中不存在@Subscribe注解的方法即时会抛出该异常，这表示在Subscriber中必须声明SubscriberMethod。
findUsingInfo
/**
* 获取Subscriber中的事件方法
*
* @param subscriberClass
* @return
*/
private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
// 1、缓存类，将一些相关的属性封装
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
// 3.0特性，通过编译器生成的索引获取事件方法
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 {
// 如果没有配置注解，就会通过反射获取事件方法，最后被存储到(findState.subscriberMethods)
findUsingReflectionInSingleClass(findState);
}
// 寻找父类，如果不存在父类，退出循环
findState.moveToSuperclass();
}
return getMethodsAndRelease(findState);
}
在上面方法中，有一个FindState类，该类是一个静态类，主要的作用就是存储Subscriber中SubscriberMethod。
通过上面第24行代码，通过调用findUsingReflectionInSingleClass()方法，将所有的SubscriberMethod都存储到findState中，最有通过调用getMethodsAndRelease()获取findState中缓存的SubscriberMethod。
findUsingReflectionInSingleClass
这个方法的作用就是通过反射获取SubscriberMethod，具体的逻辑在下面的代码中都有注释，这里就不多做处理了。
/**
* 通过反射获取类中的事件方法
*
* @param findState
*/
private void findUsingReflectionInSingleClass(FindState findState) {
Method[] methods;
try {
// This is faster than getMethods, especially when subscribers are fat classes like Activities
// 获取所有的方法
methods = findState.clazz.getDeclaredMethods();
} catch (Throwable th) {
// Workaround for java.lang.NoClassDefFoundError, see https://github.com/greenrobot/EventBus/issues/149
methods = findState.clazz.getMethods();
findState.skipSuperClasses = true;
}
for (Method method : methods) {
int modifiers = method.getModifiers();
// 事件方法必须是public的，并且不能MODIFIERS_IGNORE这几个关键字修饰
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) {
Class<?> eventType = parameterTypes[0];
if (findState.checkAdd(method, eventType)) {
// 找到一个事件方法，并存储到集合中
ThreadMode threadMode = subscribeAnnotation.threadMode();
findState.subscriberMethods.add(new SubscriberMethod(method, eventType, threadMode,
subscribeAnnotation.priority(), subscribeAnnotation.sticky()));
}
}
} else if (strictMethodVerification && method.isAnnotationPresent(Subscribe.class)) {
String methodName = method.getDeclaringClass().getName() + "." + method.getName();
throw new EventBusException("@Subscribe method " + methodName +
"must have exactly 1 parameter but has " + parameterTypes.length);
}
} else if (strictMethodVerification && method.isAnnotationPresent(Subscribe.class)) {
String methodName = method.getDeclaringClass().getName() + "." + method.getName();
throw new EventBusException(methodName +
" is a illegal @Subscribe method: must be public, non-static, and non-abstract");
}
}
}
getMethodsAndRelease
/**
* 返回Event处理方法并清除findstate的状态
*
* @param findState
* @return
*/
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;
}
这是获取SubscriberMethod的最后一步，主要就是获取通过反射获取并缓存在FindState中的SubscriberMethod列表。
并且释放FindState，这里使用了一个缓冲池，主要目的不言而喻，就是为了避免FindState对象的频繁创建。
subscribe()
通过上面SubscriberMethodFinder中一系列的方法，就可以获取到Subscriber中所有的SubscriberMethod。
在register()方法中最后一步，通过调用subscribe()方法，完成Subscriber和具体SubscriberMethod关联。
在说subscribe()之前，这里说明几个EventBus中定义的静态成员变量。如下:
/**
* 存放指定EventType相关联的所有超类或接口的EventType，这个在消息继承的机制中会使用到。
* <p>
* key：EventType（事件类型）
* Value：EventType下所有超类或接口的EventType集合
*/
private static final Map<Class<?>, List<Class<?>>> eventTypesCache = new HashMap<>();
/**
* 所有使用到指定EventType的Subscription集合（Subscription：只要包括subscriber和subscriberMethod）。
* 最后在通过post(event)方法发送事件时，就是通过在该集合中找到指定类型EventType的Subscription，最后通过反射调用subscriber的subscriberMethod方法
* <p>
* key：EventType（事件类型）
* Value：EventType相关的所有Subscription
*/
private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;
/**
* 同一个Subscriber中所有的事件方法
* <p>
* <Subscriber, List<EventType>
*/
private final Map<Object, List<Class<?>>> typesBySubscriber;
/**
* 存储sticky事件
*/
private final Map<Class<?>, Object> stickyEvents;
其中集合subscriptionsByEventType和typesBySubscriber变量是处理事件方法的核心变量，这个在post()方法中会说明。
/**
* @param subscriber
* @param subscriberMethod
*/
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
// 获取事件处理方法的参数类型
Class<?> eventType = subscriberMethod.eventType;
// 创建subscriber和事件方法的关系
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
/**
* 1、相同事件类型的事件方法按照priority排序
*
* 这段代码的整体逻辑是为了
* 创建具有相同事件类型的方法的集合，并根据priority排序，在处理事件时按照优先级处理
*
* **/
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;
}
}
/**
* 2、同一个subscriber中，存储所有事件方法（创建）
*
* 存储key-value：<subscriber，List<EventType>>，存储subscriber类中所有的EventType
*/
List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
if (subscribedEvents == null) {
subscribedEvents = new ArrayList<>();
typesBySubscriber.put(subscriber, subscribedEvents);
}
subscribedEvents.add(eventType);
/**
* 3、粘性事件的处理
*
* 如果事件方法为sticky方法，那么在注册的时候需要处理sticky事件
*/
if (subscriberMethod.sticky) {
if (eventInheritance) { // 默认true
// 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>).
/**
* 如果没有通过postSticky发送event，stickyEvents为空。
*/
Set<Map.Entry<Class<?>, Object>> entries = stickyEvents.entrySet();
for (Map.Entry<Class<?>, Object> entry : entries) {
Class<?> candidateEventType = entry.getKey();
// 判断事件类型eventType和stickyEvents中的事件类型(包括超类、接口)相同
if (eventType.isAssignableFrom(candidateEventType)) {
// 存储的事件
Object stickyEvent = entry.getValue();
checkPostStickyEventToSubscription(newSubscription, stickyEvent);
}
}
} else {
Object stickyEvent = stickyEvents.get(eventType);
checkPostStickyEventToSubscription(newSubscription, stickyEvent);
}
}
}
在subscribe()方法中，主要完成了以下工作：
- 在subscriptionsByEventType集合中存储统一个EventType的所有订阅者，并按照Event的priority排序
- 在typesBySubscriber集合中存储同一个Subscriber中所有的EventType
- 对StickyEvent进行处理
发送Event
post()
发送Event事件同时也是处理Event的过程，在该流程中主要的功能就是：
通过发送的EventType获取所有使用该EventType的Subscriber，最后通过反射实现Subscriber对指定EventType方法的调用。
这个过程主要有以下方法：
post()->postSingleEvent()->postSingleEventForEventType()->postToSubscription()->invokeSubscriber()
最终通过反射完成方法的调用，由于该逻辑的代码篇幅较长，这里贴出相关代码，在代码中有具体的说明。
/**
* 发送特定事件到 event bus
*/
public void post(Object event) {
// 线程本地变量，这种做法类似Looper机制，一个线程维护一个事件集合，所以eventbus支持多线程事件处理
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;
}
}
}
/**
* 发送单一事件
*
* @param event 事件类型参数
* @param postingState
* @throws Error
*/
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
// 获取事件参数类型
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
/**
* 默认eventInheritance是true，就是需要对事件消息继承类型进行处理
*/
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));
}
}
}
/**
* @param event 事件类型参数
* @param postingState
* @param eventClass
* @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;
} finally {
postingState.event = null;
postingState.subscription = null;
postingState.canceled = false;
}
if (aborted) {
break;
}
}
return true;
}
return false;
}
/**
* 根据线程类型调用方法
*
* @param subscription
* @param event
* @param isMainThread
*/
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);
}
}
/**
* 调用subscriber类中的方法
*
* @param subscription
* @param event
*/
void invokeSubscriber(Subscription subscription, Object event) {
try {
// 通过反射调用方法
subscription.subscriberMethod.method.invoke(subscription.subscriber, event);
} catch (InvocationTargetException e) {
handleSubscriberException(subscription, event, e.getCause());
} catch (IllegalAccessException e) {
throw new IllegalStateException("Unexpected exception", e);
}
}
postSticky()
public void postSticky(Object event) {
synchronized (stickyEvents) {
stickyEvents.put(event.getClass(), event);
}
// Should be posted after it is putted, in case the subscriber wants to remove immediately
post(event);
}
发送Sticky事件，主要就是在内存中保存该Event，等到具有该Event类型的方法被订阅，也就是调用register()方法时会被处理，这个流程可以在subscribe()方法中看到具体实现。
从subscribe()方法中可以看出，只有SubscriberMethod的sticky为true时，这个方法才会被当做粘性事件被处理。
取消注册Subscriber
unregister
关于取消注册Subscriber，其实就是将上面subscriptionsByEventType和typesBySubscriber集合中存储的有关指定Subscriber的事件移除即可。这样在处理事件时该Subscriber就不会订阅方法就不会被调用了。
/**
* 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) {
unsubscribeByEventType(subscriber, eventType);
}
// 删除该subscriber
typesBySubscriber.remove(subscriber);
} else {
logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
}
}
/**
* 移除事件类型集合中的相关事件方法
* <p>
* Only updates subscriptionsByEventType, not typesBySubscriber! Caller must update typesBySubscriber.
*/
private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
// 获取
List<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions != null) {
int size = subscriptions.size();
for (int i = 0; i < size; i++) {
// 移除集合中指定的subscriber
Subscription subscription = subscriptions.get(i);
if (subscription.subscriber == subscriber) {
subscription.active = false;
subscriptions.remove(i);
i--;
size--;
}
}
}
}
这样EventBus的基本工作原理及源码源码分析就到此为止了。
