二、EventBus源码解析
目录
- 1、EventBus的构造方法
- 2、订阅者注册
- 2.1 订阅者方法的查找过程
- 2.2 订阅者的注册过程
- 1. subscriptionsByEventType 映射：
- 2. typesBySubscriber 映射：
- 2.3 总结订阅者的注册过程
- 3、事件的发送
- 3.1 使用Post提交事件
- 3.2 使用postSingleEventForEventType处理事件的分发
- 3.2 总结事件的发送过程
- 4、订阅者的取消
1、EventBus的构造方法
我们在使用Eventbus时首先会调用Eventbus.getDefault()，用于获取Eventbus实例，我们可以看见Eventbus.getDefault()使用了DCL模式，下面简单解释一个这个模式。
public static EventBus getDefault() {
EventBus instance = defaultInstance;
if (instance == null) {
synchronized (EventBus.class) {
instance = EventBus.defaultInstance;
if (instance == null) {
instance = EventBus.defaultInstance = new EventBus();
}
}
}
return instance;
}
在这里对instance进行了两次判空处理：
- 第一次判空的作用是为了减少不必要的同步开销
假设没有第一次检查，每次调用
getDefault方法时都会进入同步块，即使实例已经被创建。这会导致在多线程环境中，多个线程频繁地竞争同步块，造成性能开销。
- 第二次判空的作用是为了防止多次创建实例
假设没有第二次检查，那么在进入同步块之前，如果有多个线程同时通过了第一次检查，它们都会进入同步块，然后按顺序创建实例。这样就违反了单例模式的初衷，因为会创建多个实例。
注意这句话：instance = EventBus.defaultInstance;
这个双重检查的模式是为了保证在高并发环境下仍能正确实现懒加载的单例模式。虽然在某些情况下可能看起来多余，但是在并发编程中，确保正确性是至关重要的。
接下来看Eventbus构造方法做了什么事情:
public EventBus() {
this(DEFAULT_BUILDER);
}
它使用了一个默认的构造器来构造Eventbus
private static final EventBusBuilder DEFAULT_BUILDER = new EventBusBuilder();
这里的this通过调用另一个Eventbus的构造方法使用了建造者模式来创建
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;
}
2、订阅者注册
首先我们明确四个名词的关系：
2.1 订阅者方法的查找过程
当获取Eventbus以后就可以将订阅者注册到Eventbus中了，接下来看一下register方法：
public void register(Object subscriber) {
if (AndroidDependenciesDetector.isAndroidSDKAvailable() && !AndroidDependenciesDetector.areAndroidComponentsAvailable()) {
// Crash if the user (developer) has not imported the Android compatibility library.
throw new RuntimeException("It looks like you are using EventBus on Android, " +
"make sure to add the \"eventbus\" Android library to your dependencies.");
}
Class<?> subscriberClass = subscriber.getClass();
//1、
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
//2、
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}
我们可以看出register做了两件事，一件是查找订阅者的订阅方法，另一件事是订阅者的注册。
在第一件事中，SubscribeMethod类中，主要用来保存Method对象，线程模式、事件类型、优先级、是否为黏性事件等等，接下来我们看一下findSubscribeMethod方法。
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
//1、
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}
//2、
if (ignoreGeneratedIndex) {
//使用反射的方法
subscriberMethods = findUsingReflection(subscriberClass);
} else {
//使用索引的方法查找
subscriberMethods = findUsingInfo(subscriberClass);
}
//3、
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;
}
}
第一步首先在缓存中查找是否存在订阅方法的集合，如果找到了直接返回即可。
第二步是根据ignoreGeneratedIndex的属性选择用何种方法查找订阅集合，
ignoreGeneratedIndex的默认值是false，使用索引的方式用于更高效地查找订阅者方法。
第三步是将找的的订阅集合放入缓存(METHOD_CACHE)中，以免下次继续查找。
顺便说一下这个缓存是什么，这个缓存是一个Map：
private static final Map<Class<?>, List<SubscriberMethod>> METHOD_CACHE = new ConcurrentHashMap<>();
ConcurrentHashMap** 是 Java 中的一个特殊的 Map 实现，它提供了一种线程安全的方式来存储键值对。**它主要采用了分段锁（Segment）的机制。其核心思想是将整个数据结构分成多个独立的段，每个段上都有一个独立的锁。这样，不同的线程可以同时访问不同的段，从而提高并发性能。
📌我们在项目中经常使用EventBus单例模式获取默认的EventBus对象，也就是
ignoreGeneratedIndex为fasle的情况，这种情况就是调用了索引的方法查找
此时我们分析索引查找的这个findUsingInfo()方法：
private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
//1、获取订阅者信息
findState.subscriberInfo = getSubscriberInfo(findState);
if (findState.subscriberInfo != null) {
//2、得到订阅方法的相关信息
SubscriberMethod[] array = findState.subscriberInfo.getSubscriberMethods();
for (SubscriberMethod subscriberMethod : array) {
if (findState.checkAdd(subscriberMethod.method, subscriberMethod.eventType)) {
findState.subscriberMethods.add(subscriberMethod);
}
}
} else {
//3、使用反射的方法查找，将信息放入findState中
findUsingReflectionInSingleClass(findState);
}
findState.moveToSuperclass();
}
//对findState做回收处理并返回订阅方法的List集合
return getMethodsAndRelease(findState);
}
我们对源码分析可以看出来findUsingInfo主要做了三件事。
- 第一件事是获取了获取了订阅者的信息，
FindState是这个SubscriberMethodFinder的内部类，包含了订阅者的信息。 - 第二件事是获取了订阅方法的相关信息，获取了包含订阅方法信息的数组，然后遍历数组存入
findState中。 - 第三件事是如果订阅者信息没有正常获取那么则通过反射的方法查找，这个具体实现在后面会介绍
当完成这三件事情以后就可以返回订阅方法的list集合，在返回之前先注销了订阅者。
现在我们看一下findUsingReflectionInSingleClass方法做了什么事情。
private void findUsingReflectionInSingleClass(FindState findState) {
Method[] methods;
try {
// This is faster than getMethods, especially when subscribers are fat classes like Activities
//1、通过反射的方法获取订阅者中的所有方法
methods = findState.clazz.getDeclaredMethods();
} catch (Throwable th) {
// Workaround for java.lang.NoClassDefFoundError, see https://github.com/greenrobot/EventBus/issues/149
try {
methods = findState.clazz.getMethods();
} catch (LinkageError error) { // super class of NoClassDefFoundError to be a bit more broad...
String msg = "Could not inspect methods of " + findState.clazz.getName();
if (ignoreGeneratedIndex) {
msg += ". Please consider using EventBus annotation processor to avoid reflection.";
} else {
msg += ". Please make this class visible to EventBus annotation processor to avoid reflection.";
}
throw new EventBusException(msg, error);
}
findState.skipSuperClasses = true;
}
for (Method method : methods) {
int modifiers = method.getModifiers();
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];
//2、保存订阅方法
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");
}
}
}
在最上面的注释1中，通过反射的方法获取了订阅者的所有方法，然后根据方法的类型，参数、注解来找到订阅方法。在注释2中将找到的订阅方法保存在分findState中。
2.2 订阅者的注册过程
在查找完订阅者的订阅方法以后，对所有的订阅方法进行注册。使用流程图理解这个方法的运行过程：
subscribe源码分析如下：
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
//获取方法的事件类型
Class<?> eventType = subscriberMethod.eventType;
//1、根据订阅者信息和订阅者方法创建一个订阅对象
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
//2、获取订阅对象的集合
CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions == null) { //如果订阅对象集合为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) {
//3、更具订阅方法的优先级插入到订阅对象集合中完成注册
subscriptions.add(i, newSubscription);
break;
}
}
//4、subscribedEvents(事件类型集合)
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);
}
}
}
我们更具四个注解对这个方法进行分析：
- 在注释1通过subscriber(订阅者信息)和SubscriberMethod(订阅者方法)创建一个订阅对象
- 在注释2中根据eventType(方法事件类型)获取subscriptions(订阅对象集合），如果订阅对象集合为null则重新创建集合并保存到subscriptionsByEventType
- 在注释3中根据订阅方法的优先级插入到订阅对象集合中完成注册
- 在注解4中根据subscriber获取subscribedEvents(事件类型集合），如果为事件类型集合null则重新创建并保存到typesBySubscriber，接下来eventType(方法事件类型)添加到subscribedEvents(事件类型集合）
如果是黏性事件则从stickyEvents事件保存队列中取出该事件类型发送给当前订阅者。
总结来说这个方法做了两件事，第一件事情是将subscriptions根据eventType封装到subscriptionsByEventType中，将subscribedEvents根据subscriber封装到typesBySubscriber中。
📌subscriptionsByEventType与typesBySubscriber的作用？
1. subscriptionsByEventType 映射：
- 类型：
Map<Class<?>, CopyOnWriteArrayList<Subscription>> - 作用：
- 维护事件类型到订阅者列表的映射。
- 允许快速查找对特定事件类型感兴趣的订阅者。
- 详细说明：
- 这个映射的键是事件类型（
Class<?>表示）——eventType = subscriberMethod.eventType。 - 值是
CopyOnWriteArrayList<Subscription>，每个Subscription包含了订阅该事件类型的订阅者的相关信息——subscriptions。
- 这个映射的键是事件类型（
- 用途：
- 在事件发布时，通过这个映射可以快速找到对应事件类型的订阅者，以便通知它们处理事件。
2. typesBySubscriber 映射：
- 类型：
Map<Object, List<Class<?>>> - 作用：
- 维护订阅者到其关注的事件类型列表的映射。
- 允许快速检索特定订阅者感兴趣的事件类型。
- 详细说明：
- 这个映射的键是订阅者对象（
Object表示）——subscriber。 - 值是
List<Class<?>>，包含了订阅者关注的事件类型——subscribedEvents。
- 这个映射的键是订阅者对象（
- 用途：
- 在订阅者注册和取消注册时，通过这个映射可以快速查找订阅者关注的事件类型，以便更新订阅者的事件类型列表。
总的来说，typesBySubscriber就是用来管理订阅机制、subscriptionsByEventType用于管理事件发送机制。
2.3 总结订阅者的注册过程
3、事件的发送
3.1 使用Post提交事件
在获取EventBus对象以后，可以通过post方法对事件进行提交。可以先通过流程图了解一下Post具体是做了什么
post源码如下：
public void post(Object event) {
//PostingThreadState保存事件队列和线程状态信息
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;
}
}
}
首先从PostingThreadState对象中取出事件队列，然后将当前事件放入事件队列中。最后将队列中的事件依次交由postSingleEvent方法处理，并移除该事件。
接下来查看postSingleEvent方法做了什么：
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));
}
}
}
eventInheritance表示是否向上查找事件的父类，默认为true,可以通过EventBusBuider进行配置。
当eventInheritance为true时，使用lookupAllEventTypes找到父类的所有事件，将这些事件放入一个List中，然后通过postSingleEventForEventType方法对事件逐一处理。
📌为什么要向上查找事件的父类？
查找事件类型的所有父类是为了支持事件类型的继承关系。在事件总线系统中，有时候我们可能定义了一些事件类型的继承关系。这种情况下，如果某个订阅者订阅了父类的事件，那么它也应该能够接收到子类的事件。
3.2 使用postSingleEventForEventType处理事件的分发
现在我们分析一下postSingleEventForEventType方法是如何处理每一个事件的。
首先通过流程图分析方法思路：
通过postSingleEventForEventType源码分析：
private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
CopyOnWriteArrayList<Subscription> subscriptions;
synchronized (this) {
//1、取出事件对应的subscriptions(订阅对象集合)
subscriptions = subscriptionsByEventType.get(eventClass);
}
if (subscriptions != null && !subscriptions.isEmpty()) {
//2、遍历subscriptions
for (Subscription subscription : subscriptions) {
postingState.event = event;
postingState.subscription = subscription;
boolean aborted;
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;
}
首先在注释1的同步块中取出该事件对应的Subscriptions（订阅对象集合）。然后在注释2位置遍历Subscriptions，将事件Event和Subscription（订阅对象）传递给postingState并调用postToSubscription方法对事件处理
接下来我们查看postToSubscription方法做了什么事情，这个方法做的事情就很简单了一个Switch语句处理不同线程状态，流程图如下：
然后我们现在看一下源码是怎么样做的：
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);
}
}
取出订阅者的threadMode线程以后，根据不同的threadMode分别处理。如果是MAIN线程则直接通过反射运行订阅方法，如果不是主线程则需要mainThreadPoster添加到主线程队列中。
mainThreadPoster是HandlerPoster类型的，继承自Handler，通过Handler调用订阅方法切换到主线程执行。
3.2 总结事件的发送过程
4、订阅者的取消
订阅者的注销需要使用到unregister方法。如下：
public synchronized void unregister(Object subscriber) {
//1、通过subscriber找到事件类型集合
List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
if (subscribedTypes != null) {
//2、遍历subscribedTypes事件类型集合，并且调用unsubscribeByEventType
for (Class<?> eventType : subscribedTypes) {
unsubscribeByEventType(subscriber, eventType);
}
//3、移除对应的subscriber对应的eventType
typesBySubscriber.remove(subscriber);
} else {
logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
}
}
这里用到了在注册中使用的typesBySubscriber，这是一个Map集合。在注释1找到这个事件类型集合，然后在注释2遍历事件类型集合，调用unsubscribeByEventType。最后在注释3将subscriber对应的eventType。
我们看一下注释2的unsubscribeByEventType方法做了什么事情：
private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
//1、通过eventType获取对应的subscriptions（订阅对象集合）
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--;
}
}
}
}
通过eventType获取对应的subscriptions（订阅对象集合）,通过一个for移除对应的subscriber即可。
