一、要学习EventBus源码一定要了解的是观察者模式。本文适合对EventBus熟悉的开发者阅读
https://www.jianshu.com/p/110cc592226e
二、EventBus的构造方法先看起来
//EventBusBuilder 从builder中初始化一些变量，Builder模式很简单，不明白的google
EventBus(EventBusBuilder builder) {
logger = builder.getLogger();
// private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;
//subscriptionsByEventType是一个key为订阅者的字节码文件，value是一个存储着订阅者本身对象和一个SubscriberMethod对象（订阅方法的所有信息）
//Subcription类里面有subscriber（订阅者）SubscriberMethod（订阅方法相关的信息）
subscriptionsByEventType = new HashMap<>();
//private final Map<Object, List<Class<?>>> typesBySubscriber;
//key是订阅者，value是一个当前订阅者的所有要监听的事件类型字节码集合
//它用来判断是否注册过eventbus和unregister（解绑evnetbus）
typesBySubscriber = new HashMap<>();
//stickyEvents key为当前的事件的.class对象，value是当前事件
stickyEvents = new ConcurrentHashMap<>();
/**
private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {
@Override
protected PostingThreadState initialValue() {
return new PostingThreadState();
}
};
**/
//mainThreadSupport是一个ThreadLocal（不清楚这个的可以自己去看看，
//简单说一下就是每个线程都独自管理PostingThreadState，也就是线程和PostingThreadState一对一）
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是一个获取订阅者的所有订阅方法的类，主要包括两种方式，一种是通过反射整个类，拿到带有@Subscribe注解的方法。还有一种是通过注解器为我们生成的索引MyEventBusIndex来查找对应的订阅方法。
subscriberMethodFinder = new SubscriberMethodFinder(builder.subscriberInfoIndexes,
builder.strictMethodVerification, builder.ignoreGeneratedIndex);
logSubscriberExceptions = builder.logSubscriberExceptions;
logNoSubscriberMessages = builder.logNoSubscriberMessages;
sendSubscriberExceptionEvent = builder.sendSubscriberExceptionEvent;
sendNoSubscriberEvent = builder.sendNoSubscriberEvent;
throwSubscriberException = builder.throwSubscriberException;
//表示是否向上查找事件的父类,默认为true
eventInheritance = builder.eventInheritance;
//Executors.newCachedThreadPool() 没有核心线程，可无限创建非核心线程的类
executorService = builder.executorService;
}
三、看主流程，首先看一下EvnetBus的注册
EventBus.getDefault().register(this);
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;
}
就是一个双doule check的单列
在看看register源码
public void register(Object subscriber) {
Class<?> subscriberClass = subscriber.getClass();
//查找所有的订阅的方法
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
//把订阅者放到对应的集合中subscriptionsByEventType
subscribe(subscriber, subscriberMethod);
}
}
}
分析下如何findSubscriberMethods（）
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
//首先是从缓存中拿订阅者所有订阅的方法集合
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}
//ignoreGeneratedIndex 是否忽略索引，忽略索引会通过反射拿到对应的@Subscriber对应的信息，
//开启索引则是通过注解解析器来获取订阅回调方法信息
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;
}
}
找到通过反射获取注解信息源码 findUsingReflection（） # findUsingReflectionInSingleClass（）
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的，并且不是static 和abstract的
if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
//获取Class类型的数组
Class<?>[] parameterTypes = method.getParameterTypes();
if (parameterTypes.length == 1) {
//获取带Subscribe的注解信息
Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
if (subscribeAnnotation != null) {
Class<?> eventType = parameterTypes[0];
//检查事件是不是唯一和签名是否完整，并且同一个订阅者不能有多个的方法监听同一个事件
if (findState.checkAdd(method, eventType)) {
ThreadMode threadMode = subscribeAnnotation.threadMode();
//拿到订阅方法的信息，封装成SubscriberMethod，然后存到findState中
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");
}
}
看一下findUsingInfo（）这种方式获取所有订阅方法信息的源码
private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
//查看注解器为我们生产的索引MyEventBusIndex，如果配置了就会获取到对应的SubscriberInfo信息，也就是一些注解信息，
//如果没有配置就直接使用注解了
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 {
findUsingReflectionInSingleClass(findState);
}
findState.moveToSuperclass();
}
return getMethodsAndRelease(findState);
}
继续看下上面的最后一行getMethodsAndRelease（findState）；
//从findState里取出了subscriberMethods
List<SubscriberMethod> subscriberMethods = new ArrayList<>(findState.subscriberMethods);
//回收findstate
findState.recycle();
//把findState缓存到FindState池中，方便下一次使用，提高性能。
synchronized (FIND_STATE_POOL) {
for (int i = 0; i < POOL_SIZE; i++) {
if (FIND_STATE_POOL[i] == null) {
FIND_STATE_POOL[i] = findState;
break;
}
}
}
//返回对应的subscriberMethods信息
return subscriberMethods;
四、register的findSubscriberMethods我们看完了，接下来继续看下subscribe(subscriber, subscriberMethod);源码
// Must be called in synchronized block
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
Class<?> eventType = subscriberMethod.eventType;
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
//根据事件类型来查找所有订阅了这个事件的订阅信息
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;
}
}
//typesBySubscriber这个主要是用来判断是否注册的集合，这里注册添加进去。
List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
if (subscribedEvents == null) {
subscribedEvents = new ArrayList<>();
typesBySubscriber.put(subscriber, subscribedEvents);
}
subscribedEvents.add(eventType);
//判断是否有黏性事件，（粘性事件大家应该都知道，就是可以先发送消息，后去订阅，也可以收到之前的消息）其实原理很简单
//，在postSticky的时候把发送的类型和对应的事件都缓存起来，然后在register的时候找到之前的缓存，重新发送一次消息，
//然后就会回调事件
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);
}
}
}
二、来看下EventBus的post事件
/** Posts the given event to the event bus. */
public void post(Object event) {
//根据当前线程获取对应的evnetQueue，以及线程线程信息等，在下面会贴出这个类
PostingThreadState postingState = currentPostingThreadState.get();
List<Object> eventQueue = postingState.eventQueue;
//把当前post的event放到事件队列里面
eventQueue.add(event);
//判断当前是否处于投递状态
if (!postingState.isPosting) {
//如果没有在投递，给postingState赋值是否是主线程
postingState.isMainThread = isMainThread();
//设置为投递状态
postingState.isPosting = true;
if (postingState.canceled) {
throw new EventBusException("Internal error. Abort state was not reset");
}
try {
//判断当前投递事件队列是否是空，不为空就执行postSingleEvent
while (!eventQueue.isEmpty()) {
postSingleEvent(eventQueue.remove(0), postingState);
}
} finally {
//投递结束，复位标志位
postingState.isPosting = false;
postingState.isMainThread = false;
}
}
}
分析下postSingleEvent(eventQueue.remove(0), postingState);源码
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
//eventInheritance默认是true，在发射事件的时候会判断是否需要发射父类事件
if (eventInheritance) {
//获取事件类型，并在eventTypesCache中做了缓存
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));
}
}
}
继续看下postSingleEventForEventType()源码
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;
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(subscription, event, postingState.isMainThread);源码
private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
//判断订阅者所在的线程，根据5种不同的线程模式来切换到制定的线程来回调方法
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);
}
}
1、POSTING：执行 invokeSubscriber() 方法，内部直接采用反射调用。
2、MAIN：首先去判断当前是否在 UI 线程，如果是的话则直接反射调用，否则调用mainThreadPoster的enqueue()方法，即把当前的方法加入到队列之中，然后通过 handler 去发送一个消息，在 handler 的 handleMessage 中去执行方法。
3、MAIN_ORDERED：与MAIN类似，不过是确保是顺序执行的。
4、BACKGROUND：判断当前是否在 UI 线程，如果不是的话则直接反射调用，是的话通过backgroundPoster的enqueue()方法 将方法加入到后台的一个队列，最后通过线程池去执行。注意，backgroundPoster在 Executor的execute()方法 上添加了 synchronized关键字 并设立 了控制标记flag，保证任一时间只且仅能有一个任务会被线程池执行。
5、ASYNC：逻辑实现类似于BACKGROUND，将任务加入到后台的一个队列，最终由Eventbus 中的一个线程池去调用，这里的线程池与 BACKGROUND 逻辑中的线程池用的是同一个，即使用Executors的newCachedThreadPool()方法创建的线程池，它是一个有则用、无则创建、无数量上限的线程池。不同于backgroundPoster的保证任一时间只且仅能有一个任务会被线程池执行的特性，这里asyncPoster则是异步运行的，可以同时接收多个任务。
网友评论