EventBus源码解析(一)
2018-08-01 本文已影响0人
猫KK
上一篇我们介绍了EventBus的简单使用，但是对于一个库，不能简单的会用就行，我们得了解里面的原理。
问题
看源码我们不能硬看，得带着问题看。对于EventBus我们带着两个问题去了解
1.发送消息后是怎么能执行我们注解的方法
2.怎么根据ThreadMode来切换线程
第一怎么回调
我们先来看看这个方法
EventBus.getDefault().register(this);
点击去看getDefault
public static EventBus getDefault() {
if (defaultInstance == null) {
synchronized (EventBus.class) {
if (defaultInstance == null) {
defaultInstance = new EventBus();
}
}
}
return defaultInstance;
}
简单单例，创建EventBus 对象
在看register 方法
public void register(Object subscriber) {
//获取我们传进去的Class
Class<?> subscriberClass = subscriber.getClass();
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}
继续看findSubscriberMethods()方法实现
private static final Map<Class<?>, List<SubscriberMethod>> METHOD_CACHE = new ConcurrentHashMap<>();
SubscriberMethodFinder(List<SubscriberInfoIndex> subscriberInfoIndexes, boolean strictMethodVerification,
boolean ignoreGeneratedIndex) {
this.subscriberInfoIndexes = subscriberInfoIndexes;
this.strictMethodVerification = strictMethodVerification;
//创建的时候直接传过来
this.ignoreGeneratedIndex = ignoreGeneratedIndex;
}
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
//缓存策略，如果有就直接使用，没有就新建，第一次绑定肯定是没有
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}
//我们需要看这个参数的值，查看构造函数
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;
}
}
回到register()方法中查看SubscriberMethodFinder什么时候被创建
//直接创建EventBusBuilder 对象
private static final EventBusBuilder DEFAULT_BUILDER = new EventBusBuilder();
public EventBus() {
//前面提到使用单例创建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;
}
同志们，看到这里就应该知道使用了Builder 设计模式，用来构建参数
查看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 默认为false
boolean ignoreGeneratedIndex;
boolean strictMethodVerification;
ExecutorService executorService = DEFAULT_EXECUTOR_SERVICE;
List<Class<?>> skipMethodVerificationForClasses;
List<SubscriberInfoIndex> subscriberInfoIndexes;
Logger logger;
MainThreadSupport mainThreadSupport;
EventBusBuilder() {
}
........
/** Forces the use of reflection even if there's a generated index (default: false). */
public EventBusBuilder ignoreGeneratedIndex(boolean ignoreGeneratedIndex) {
this.ignoreGeneratedIndex = ignoreGeneratedIndex;
return this;
}
.......
}
让我们回到findSubscriberMethods()方法中去，我们知道ignoreGeneratedIndex 默认为false，查看findUsingReflection()的实现
private static final FindState[] FIND_STATE_POOL = new FindState[POOL_SIZE];
private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
//获取FindState 对象，用来保存内容
FindState findState = prepareFindState();
//初始化findState
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
//获取方法
findUsingReflectionInSingleClass(findState);
findState.moveToSuperclass();
}
return getMethodsAndRelease(findState);
}
//如果数组中存在，就从数组中拿，否则就创建一个
private FindState prepareFindState() {
synchronized (FIND_STATE_POOL) {
for (int i = 0; i < POOL_SIZE; i++) {
FindState state = FIND_STATE_POOL[i];
if (state != null) {
FIND_STATE_POOL[i] = null;
return state;
}
}
}
return new FindState();
}
static class FindState {
final List<SubscriberMethod> subscriberMethods = new ArrayList<>();
final Map<Class, Object> anyMethodByEventType = new HashMap<>();
final Map<String, Class> subscriberClassByMethodKey = new HashMap<>();
final StringBuilder methodKeyBuilder = new StringBuilder(128);
Class<?> subscriberClass;
Class<?> clazz;
boolean skipSuperClasses;
SubscriberInfo subscriberInfo;
void initForSubscriber(Class<?> subscriberClass) {
this.subscriberClass = clazz = subscriberClass;
skipSuperClasses = false;
subscriberInfo = null;
}
}
接下来看主要的findUsingReflectionInSingleClass()方法
private void findUsingReflectionInSingleClass(FindState findState) {
Method[] methods;
try {
// This is faster than getMethods, especially when subscribers are fat classes like Activities
//findState其实就是我们传进来的this，获取this类的所有方法
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();
//判断方法的修饰符，不是private修饰的方法
if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
//获取方法的参数
Class<?>[] parameterTypes = method.getParameterTypes();
//参数的长度为 1
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 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)) {
//private 修饰的方法直接跑异常
String methodName = method.getDeclaringClass().getName() + "." + method.getName();
throw new EventBusException(methodName +
" is a illegal @Subscribe method: must be public, non-static, and non-abstract");
}
}
}
到此EventBus.getDefault().register(this);已经算走完了。这个方法主要就是将该类中带有Subscribe 注解的方法找到并保存，接下来我们查看发送消息的方法
EventBus.getDefault().post(bean);
EventBus.getDefault()不用说，就是是获取单EventBus对象，主要来看post 方法实现
//使用ThreadLocal 来创建复制PostingThreadState
private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {
@Override
protected PostingThreadState initialValue() {
return new PostingThreadState();
}
};
//主要用来保存一些参数
final static class PostingThreadState {
final List<Object> eventQueue = new ArrayList<>();
boolean isPosting;
boolean isMainThread;
Subscription subscription;
Object event;
boolean canceled;
}
public void post(Object event) {
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 {
//遍历消息队列，所以多次调用post 方法，就会收到多次
while (!eventQueue.isEmpty()) {
//循环调用方法
postSingleEvent(eventQueue.remove(0), postingState);
}
} finally {
postingState.isPosting = false;
postingState.isMainThread = false;
}
}
}
下面我们来看postSingleEvent()的实现
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
//这个event 是post 是传进来的参数
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
//查看builder 就可以知道默认为true
if (eventInheritance) {
List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
int countTypes = eventTypes.size();
for (int h = 0; h < countTypes; h++) {
Class<?> clazz = eventTypes.get(h);
//主要调用postSingleEventForEventType 方法
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 = 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()方法的实现
private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
//subscription 就是我们调用register 时保存的，根据threadMode来判断调用不同的方法
switch (subscription.subscriberMethod.threadMode) {
case POSTING:
//我们现在只来分析默认的POSTING
//查看invokeSubscriber实现
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);
}
}
查看invokeSubscriber()方法的实现
void invokeSubscriber(Subscription subscription, Object event) {
try {
//很简单，根据register时找到的方法反射调用该方法event 就是post是传进来的惨数
subscription.subscriberMethod.method.invoke(subscription.subscriber, event);
} catch (InvocationTargetException e) {
handleSubscriberException(subscription, event, e.getCause());
} catch (IllegalAccessException e) {
throw new IllegalStateException("Unexpected exception", e);
}
}
到此 EventBus.getDefault().post(bean);方法也结束了，我们也清楚了知道问题一的原因，其中很多细节没有细看，我们主要看大致的实现流程，细节大家可以自己看源码
总结
EventBus.getDefault().register(this);方法就是将该类中的带有Subscribe 注解的方法找到并保存下来
EventBus.getDefault().post(bean);方法就是将前面保存下来的方法反射执行，实现组件间通信
至于问题二，如何实现线程间调度，请看下一篇
