EventBus
2020-05-27 本文已影响0人
要学的东西太多了
组件间通信除了Arouter，EventBus常用来发布/订阅事件，类似观察者模式，核心原理是注解和反射，基本使用比较简单，如下：
订阅事件的对象：
//注册，一般放在onCreate里面
EventBus.getDefault().register(this);
//接收事件的方法
@Subscribe(threadMode = ThreadMode.MAIN)
public void receiveMsg(EventMsg msg){
if(msg.getType()==0){
ToastUtils.showShortCenter(this,msg.getMessage());
}
}
//取消注册，一般在onDestroy里面
EventBus.getDefault().unregister(this);
发布事件的对象：
EventBus.getDefault().post(new EventMsg(0,msg));
//粘性事件，这种情况是订阅者还没注册的情况
EventBus.getDefault().postSticky(new EventMsg(0,msg));
接下来，进行原理分析。先看下Subscribe注解的实现：
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface Subscribe {
ThreadMode threadMode() default ThreadMode.POSTING;//方法的线程模型
boolean sticky() default false;//是否是粘性事件，默认不是
int priority() default 0;//优先级，越大越先执行
}
然后，看一下订阅的实现：
EventBus.getDefault()方法是返回一个单例对象，在EventBus构造方法里面，直接调用了this(DEFAULT_BUILDER)。
DEFAULT_BUILDER是一个默认的EventBusBuilder，这里用了构建者模式，进行了一些初始化操作：
EventBus(EventBusBuilder builder) {
logger = builder.getLogger();
//这个map是重点，类型是Map<Class<?>, CopyOnWriteArrayList<Subscription>>，用来存 事件class - 封装了订阅者、方法的Subscription对象列表，CopyOnWriteArrayList是线程安全的
subscriptionsByEventType = new HashMap<>();
//这个map的类型是Map<Object, List<Class<?>>>，存储的是订阅者和事件class列表的对应关系，用来取消注册时，根据订阅者遍历事件class列表，进而删除subscriptionsByEventType 对应的方法
typesBySubscriber = new HashMap<>();
//这个map存放的是粘性事件，类型是Map<Class<?>, Object>，key是事件class，值就是事件本身
stickyEvents = new ConcurrentHashMap<>();
//builder的这个对象持有主线程的looper
mainThreadSupport = builder.getMainThreadSupport();
//通过tMainThreadSupport的looper对象，创建一个handler
mainThreadPoster = mainThreadSupport != null ? mainThreadSupport.createPoster(this) : null;
//BackgroundPoster和AsyncPoster都实现了Runnable接口，里面都有PendingPostQueue，并且在enqueue方法会入队，并且用eventBus的executorService执行任务，在run方法里面调用eventBus.invokeSubscriber(pendingPost)，通过反射调用对应的方法。
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;
//线程池，默认是newCachedThreadPool，我们可以自己指定
executorService = builder.executorService;
}
接下来，看下register方法做了什么：
public void register(Object subscriber) {
Class<?> subscriberClass = subscriber.getClass();
//找到目标对象的方法列表
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
synchronized (this) {
//循环订阅
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}
findSubscriberMethods方法实现如下：
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
//先从缓存里面取，取到了直接返回
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}
//这个标志默认是false，所以我们重点关注findUsingInfo方法
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方法主要关注以下代码：
private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
//...
findUsingReflectionInSingleClass(findState);
//会去找父类的方法
findState.moveToSuperclass();
}
//返回findState的SubscriberMethod集合
return getMethodsAndRelease(findState);
}
findUsingReflectionInSingleClass方法通过反射找到目标对象符合条件的方法：
private void findUsingReflectionInSingleClass(FindState findState) {
Method[] methods;
try {
methods = findState.clazz.getDeclaredMethods();
} catch (Throwable th) {
try {
methods = findState.clazz.getMethods();
} catch (LinkageError error) {
//...
}
findState.skipSuperClasses = true;
}
for (Method method : methods) {
//方法修饰符
int modifiers = method.getModifiers();
//只判断是public，且不是abstract和static的方法
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();
//新建一个SubscriberMethod对象，包含方法对象、参数类型、线程模型、优先级、是否是粘性事件，并添加到列表
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");
}
}
}
register找到SubscriberMethod集合后，会循环调用subscribe方法来添加订阅，接下来分析一下subscribe方法：
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
//订阅的事件类型
Class<?> eventType = subscriberMethod.eventType;
//将订阅者和subscriberMethod封装成Subscription对象
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions == null) {
subscriptions = new CopyOnWriteArrayList<>();
subscriptionsByEventType.put(eventType, subscriptions);
} else {
//这里的目的是只能调用一次register，再调用会抛异常
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;
}
}
//这里保存订阅者和事件类型列表的关联
List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
if (subscribedEvents == null) {
subscribedEvents = new ArrayList<>();
typesBySubscriber.put(subscriber, subscribedEvents);
}
subscribedEvents.add(eventType);
//处理粘性事件，如果方法是粘性的，且粘性事件map里面有事件未处理，就反射处理
if (subscriberMethod.sticky) {
if (eventInheritance) {
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);
}
}
}
至此，订阅就完成了，unregister就是一些对象回收、清除、资源回收的工作。然后再来看下怎么发送事件的：
先分析post方法：
public void post(Object event) {
//currentPostingThreadState是ThreadLocal对象，保证每个线程的局部内存数据
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;
}
}
}
重点看一下postSingleEvent里面调用的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方法里面会根据线程模型来执行事件处理，最后都会调用invokeSubscriber方法，这里就是反射调用方法：
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方法就比较简单了，stickyEvents里面存储一个事件，并调用post方法尝试发送事件：
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);
}
