Android EventBus3.1.1从使用到源码解析
前言:EventBus相信或多或少都使用过,他可以简化组件间的通信,发布和订阅事件总线很好的分离了发布者和接收者(订阅者),发布和订阅者之间没有联系他们通过Eventbus来通信,下面看下如何使用,用过的可以直接忽略去看源码分析 本篇文章基于:compile 'org.greenrobot:eventbus:3.1.1'github:https://github.com/...
·
前言:
EventBus相信或多或少都使用过,他可以简化组件间的通信,发布和订阅事件总线很好的分离了发布者和接收者(订阅者),发布和订阅者之间没有联系他们通过Eventbus来通信,下面看下如何使用,用过的可以直接忽略去看源码分析
本篇文章基于:
compile 'org.greenrobot:eventbus:3.1.1'
github:https://github.com/greenrobot/EventBus
使用:
MainActivity
@Override
protected void onStop() {
super.onStop();
EventBus.getDefault().unregister(this);
Log.d(TAG, "onStop : ");
}
@Override
protected void onStart() {
super.onStart();
EventBus.getDefault().register(this);
Log.d(TAG, "onStart : ");
}
@Subscribe(threadMode = ThreadMode.MAIN)
public void onMessageEvent1(String text) {
Log.d(TAG, "onMessageEvent1 : String");
}
@Subscribe(threadMode = ThreadMode.BACKGROUND)
public void onMessageEvent2(String text) {
Log.d(TAG, "onMessageEvent2 : String" );
}
@Subscribe(threadMode = ThreadMode.POSTING)
public void onMessageEvent3(String text) {
Log.d(TAG, "onMessageEvent3 : String" );
}
@Subscribe(threadMode = ThreadMode.ASYNC)
public void onMessageEvent4(String text) {
Log.d(TAG, "onMessageEvent4 : String" );
}
@Subscribe(threadMode = ThreadMode.MAIN_ORDERED)
public void onMessageEvent4(Integer text) {
Log.d(TAG, "onMessageEvent4 : Integer" );
}FirstActivity
final String text = "长江长江我是黄河";
new Thread(new Runnable() {
@Override
public void run() {
EventBus.getDefault().post(text);//发送一个事件
}
}).start();
太简单都不好意思说怎么用的...
源码:
源码基于@Subscribe，threadMode，register，post，unregister这一路线来分析
@Subscribe
@Subscribe是EventBus3.0之后开始支持,3.0之前通过订阅的函数名来区分如下
-
onEvent()
-
onEventMainThread()
-
onEventBackgroundThread()
-
onEventAsync()
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface Subscribe {
ThreadMode threadMode() default ThreadMode.POSTING;
/**
* If true, delivers the most recent sticky event (posted with
* {@link EventBus#postSticky(Object)}) to this subscriber (if event available).
*/
boolean sticky() default false;
/** Subscriber priority to influence the order of event delivery.
* Within the same delivery thread ({@link ThreadMode}), higher priority subscribers will receive events before
* others with a lower priority. The default priority is 0. Note: the priority does *NOT* affect the order of
* delivery among subscribers with different {@link ThreadMode}s! */
int priority() default 0;
}@Subscribe采用运行时注解,且注解只能用在函数上,默认的threadmode为posting,sticky-粘性事件,使用如下
MainActivity
@Subscribe(threadMode = ThreadMode.MAIN,sticky = true)
public void onMessageEvent1(String text) {
Log.d(TAG, "onMessageEvent1 : String");
}FirstActivity
EventBus.getDefault().postSticky("2");//发送一个粘性事件粘性事件和普通事件的区别在于,普通事件post时如果订阅者尚未注册那么这么事件就丢失了,如果声明了sticky=true,postSticky时订阅者尚未注册那么这则消息不会丢失,在注册成功后立即就会收到
priority优先级默认为0，在通过反射获取到订阅者所有的方法时,方法是一个无序的列表,在处理这些订阅方法的时候会按照优先级来排序,但在实际post事件时经过我的测试大概会分为两种情况:
1.threadmode相同的情况下,按照优先级来排序如下
@Subscribe(threadMode = ThreadMode.MAIN_ORDERED,sticky = true,priority = 1)
public void onMessageEvent1(String text) {
Log.d(TAG, "onMessageEvent1 : priority = 1");
}
@Subscribe(threadMode = ThreadMode.MAIN_ORDERED,priority = 2)
public void onMessageEvent2(String text) {
Log.d(TAG, "onMessageEvent2 : priority = 2" );
}
@Subscribe(threadMode = ThreadMode.MAIN_ORDERED,priority = 3)
public void onMessageEvent3(String text) {
Log.d(TAG, "onMessageEvent3 : priority = 3" );
}
@Subscribe(threadMode = ThreadMode.MAIN_ORDERED,priority = 4)
public void onMessageEvent4(String text) {
Log.d(TAG, "onMessageEvent4 : priority = 4" );
}
@Subscribe(threadMode = ThreadMode.MAIN_ORDERED,priority = 5)
public void onMessageEvent5(String text) {
Log.d(TAG, "onMessageEvent4 : priority = 5" );
}
log:
priority = 5
priority = 4
priority = 3
priority = 2
priority = 12.threadmode不相同的情况下会收到线程切换.是否阻塞等影响因素,log:
priority = 3
priority = 4
priority = 1
priority = 2
priority = 5酌情使用即可....
----------------------------------------------------------不华丽的风格陷---------------------------------------------
threadMode
POSTING
默认的模式,开销最小的模式,因为声明为POSTING的订阅者会在发布的同一个线程调用，发布者在主线程那么订阅者也就在主线程,反之亦,避免了线程切换,如果不确定是否有耗时操作，谨慎使用，因为可能是在主线程发布
MAIN
主线程调用，视发布线程不同处理不同,如果发布者在主线程那么直接调用(非阻塞式),如果发布者不在主线程那么阻塞式调用,这句话怎么理解呢，看下面的Log比较清晰的理解
主线程(阻塞式):
Log.d(TAG, "run : 1");
EventBus.getDefault().post(text);//发送一个事件
Log.d(TAG, "run : 2");
EventBus.getDefault().post(text);//发送一个事件
@Subscribe(threadMode = ThreadMode.MAIN)
public void onMessageEvent1(String text) {
Log.d(TAG, "onMessageEvent1 : ");
}
日志输出
: run : 1
: onMessageEvent1 :
: run : 2
: onMessageEvent1 :
非主线程(非阻塞式):
final String text = "长江长江我是黄河";
new Thread(new Runnable() {
@Override
public void run() {
Log.d(TAG, "run : 1");
EventBus.getDefault().post(text);//发送一个事件
Log.d(TAG, "run : 2");
EventBus.getDefault().post(text);//发送一个事件
}
}).start();
日志输出:
run : 1
run : 2
onMessageEvent1 :
onMessageEvent1 :
MAIN_ORDERED
和MAIN差不多,主线程调用，和MAIN不同的是他保证了post是非阻塞式的(默认走MAIN的非主线程的逻辑，所以可以做到非阻塞)
BACKGROUND
在子线程调用,如果发布在子线程那么直接在发布线程调用,如果发布在主线程那么将开启一个子线程来调用,这个子线程是阻塞式的,按顺序交付所有事件,所以也不适合做耗时任务,因为多个事件共用这一个后台线程
ASYNC
在子线程调用,总是开启一个新的线程来调用,适用于做耗时任务,比如数据库操作,网络请求等，不适合做计算任务,会导致开启大量线程
----------------------------------------------------------不华丽的风格陷---------------------------------------------
EventBus.getDefault().register(this);
EventBus#getDefault
/** Convenience singleton for apps using a process-wide EventBus instance. */
public static EventBus getDefault() {
if (defaultInstance == null) {
synchronized (EventBus.class) {
if (defaultInstance == null) {
defaultInstance = new EventBus();
}
}
}
return defaultInstance;
}一个双重效验的单例模式,但是他又提供了public的构造函数
public EventBus() {
this(DEFAULT_BUILDER);
}通过getDefault可以获取一条默认的事件总线,通过new EventBus可以获取多条事件总线,其应用场景经过我的测试也就只能用在注册类中Post，可以实现线程切换,但是其仅仅为了线程切换而使用EventBus就太重了，所以一般情况下都是使用getDefault
EventBus#register
public void register(Object subscriber) {
Class<?> subscriberClass = subscriber.getClass();
//subscriberMethods返回的是subscriber这个类中所有的订阅方法
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);//分类保存后面有分析
}
}
}SubscriberMethodFinder#findSubscriberMethods
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
//根据订阅者去缓存中查找
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}
if (ignoreGeneratedIndex) {//默认为false
subscriberMethods = findUsingReflection(subscriberClass);
} else {
subscriberMethods = findUsingInfo(subscriberClass);
}
//如果在findUsingInfo中没有找到带有@Subscribe的public方法就抛出异常
if (subscriberMethods.isEmpty()) {
throw new EventBusException("Subscriber " + subscriberClass
+ " and its super classes have no public methods with the @Subscribe annotation");
} else {//反之就添加进缓存 并且返回这个存储了所有订阅函数的集合subscriberMethods
METHOD_CACHE.put(subscriberClass, subscriberMethods);
return subscriberMethods;
}
}大部分优秀的框架都会有自己的缓存策略,这在一定程度上减小了性能开销
SubscriberMethodFinder#findUsingInfo
private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
//主要new一个FindState返回,如果缓存池中已经存在那么不需要new 性能略好
FindState findState = prepareFindState();
//做一些初始化操作
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
//这里返回的是null 所以走else
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);
}SubscriberMethodFinder#findUsingReflectionInSingleClass
private void findUsingReflectionInSingleClass(FindState findState) {
Method[] methods;
try {
// This is faster than getMethods, especially when subscribers are fat classes like Activities
//获取订阅者的所有方法,如果发生crash就调用下面的getMethods，
//这个clazz是在initForSubscriber初始化时赋值的订阅者,按照上面的示例代码这里就是MainActivity
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
if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
Class<?>[] parameterTypes = method.getParameterTypes();
if (parameterTypes.length == 1) {//判断参数是否是一个,按照上面的实例代码,这里的parameterTypes就是String类
Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);//subscribeAnnotation中包含了优先级,黏性,threadMode信息
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()));//此处添加进subscriberMethods的中的所有订阅方法会在findUsingInfo函数中返回
}
}
} 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");
}
}
}SubscriberMethodFinder#checkAdd
boolean checkAdd(Method method, Class<?> eventType) {
// 2 level check: 1st level with event type only (fast), 2nd level with complete signature when required.
// Usually a subscriber doesn't have methods listening to the same event type.
//参数类型为Key 方法为value存储 existing是当出现重复Key时重复key的value,反之没有重复key则返回Null
//当有多个订阅方法的参数是一致的时候会走到下面的else 看下checkAddWithMethodSignature
Object existing = anyMethodByEventType.put(eventType, method);
if (existing == null) {
return true;
} else {
if (existing instanceof Method) {
//如果存在同样订阅方法参数 就会导致覆盖 所有这里还需要再put一次 那进入checkAddWithMethodSignature
if (!checkAddWithMethodSignature((Method) existing, eventType)) {
// Paranoia check
throw new IllegalStateException();
}
// Put any non-Method object to "consume" the existing Method
anyMethodByEventType.put(eventType, this);
}
return checkAddWithMethodSignature(method, eventType);
}
}SubscriberMethodFinder#checkAddWithMethodSignature
private boolean checkAddWithMethodSignature(Method method, Class<?> eventType) {
methodKeyBuilder.setLength(0);
methodKeyBuilder.append(method.getName());
methodKeyBuilder.append('>').append(eventType.getName());
String methodKey = methodKeyBuilder.toString();
Class<?> methodClass = method.getDeclaringClass();
//这里和上面的逻辑一致 methodClassOld是key重复时的重复Key的value 有点费解这话,举个例子就明白,第一次put(1, 1),返回Null,再次putput(1, 2),返回1
Class<?> methodClassOld = subscriberClassByMethodKey.put(methodKey, methodClass);
//isAssignableFrom的作用是判断methodClassOld对象所表示的类或接口是否和methodClass所表示的类或接口一致 或者是其超类
if (methodClassOld == null || methodClassOld.isAssignableFrom(methodClass)) {
// Only add if not already found in a sub class
return true;
} else {
// Revert the put, old class is further down the class hierarchy
subscriberClassByMethodKey.put(methodKey, methodClassOld);
return false;
}
}
再回到 SubscriberMethodFinder#findUsingInfo
private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
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();//为了性能超类是系统的类直接跳过这个while
}
return getMethodsAndRelease(findState);//获取到订阅方法列表且回收findState
}SubscriberMethodFinder#getMethodsAndRelease
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;//FIND_STATE_POOL是一个缓存findState的池,在前面的prepareFindState中就可以直接获取到
break;
}
}
}
return subscriberMethods;//最终返回这个subscriberMethods列表
}
再次回到findSubscriberMethods
SubscriberMethodFinder#findSubscriberMethods
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 {//根据订阅者为Key 订阅方法列表为value存储进入缓存,当下一次再次调用findSubscriberMethods函数时就可以直接根据订阅者找到这些缓存 效率略高
METHOD_CACHE.put(subscriberClass, subscriberMethods);
return subscriberMethods;
}
}再回到register
EventBus#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);
}
}
}// Must be called in synchronized block
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
Class<?> eventType = subscriberMethod.eventType;//订阅函数参数类型
//这一步很简单就是在构造函数中记录下订阅者和订阅方法
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
//CopyOnWriteArrayList是java.util包下的,他使用了写时复制的方法来实现,其效率并不高,但可以保证在多线程环境下最终(强调是最终)数据的一致性
CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);//subscriptionsByEventType可以根据参数类型来获取到订阅事件
//在操作第一个订阅事件时肯定是==null的
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);
if (subscriberMethod.sticky) {//粘性事件的处理逻辑在最后再分析,因为其内容包含了post流程
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);
}
}
}
register到此结束,如果看代码比较头疼,看总结也可以:
1.通过getDefault获取EventBus实例，这是一条默认的事件总线,通过单例模式实现,其构造函数是Public的也可以通过new的形式来创建多条事件总线
2.从缓存中获取订阅方法列表,如果缓存中不存在则通过反射获取到订阅者所有的函数,遍历再通过权限修饰符.参数长度(只允许一个参数).注解(@Subscribe)来判断是否是具备成为订阅函数的前提,具备则构建一个SubscriberMethod(订阅方法，其相当于一个数据实体类,包含方法,threadmode,参数类型,优先级,是否粘性事件这些参数),循环结束订阅函数列表构建完成添加进入缓存
3.对subscriptionsByEventType.typesBySubscriber完成数据初始化,subscriptionsByEventType根据参数类型存储订阅者和订阅方法,typesBySubscriber根据订阅者存储了所有的参数类型，subscriptionsByEventType主要是post时使用,因为其存储了订阅者和订阅事件这两个参数在反射时要用到,typesBySubscriber在反注册时可以根据订阅者获取到存储的事件类型就可以从
subscriptionsByEventType中获取到对应的订阅者和订阅方法释放资源,还可以用来判断是否注册,这些最后会分析到
----------------------------------------------------------不华丽的风格陷---------------------------------------------
EventBus.getDefault().post()
getDefault逻辑同上,看post
EventBus#post
public void post(Object event) {
//currentPostingThreadState是ThreadLocal,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;
}
}
}EventBus#postSingleEvent
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
Class<?> eventClass = event.getClass();//获取事件类型
boolean subscriptionFound = false;
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);//如果没有eventClass对应的订阅者 打印FINE信息
}
if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
eventClass != SubscriberExceptionEvent.class) {
post(new NoSubscriberEvent(this, event));
}
}
}EventBus#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;
}EventBus#postToSubscription
//这里根据是否在主线程和threadmode来判断
private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
switch (subscription.subscriberMethod.threadMode) {
case POSTING:
invokeSubscriber(subscription, event);//post在什么线程就直接调用 不需要切换线程
break;
case MAIN:
if (isMainThread) {
invokeSubscriber(subscription, event);
} else {
mainThreadPoster.enqueue(subscription, event);//如果Post在主线程直接调用,反之通过handler来切换到主线程再调用反射
}
break;
case MAIN_ORDERED:
if (mainThreadPoster != null) {//默认走这里的逻辑和MAIN一致 事件排队等待调用,非阻塞式
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 {//如果post在子线程直接在Post线程调用
invokeSubscriber(subscription, event);
}
break;
case ASYNC:
asyncPoster.enqueue(subscription, event);//总是开启线程来调用
break;
default:
throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
}
}EventBus#invokeSubscriber
void invokeSubscriber(Subscription subscription, Object event) {
try {
//这里最后说明一下subscription中包含了订阅者和订阅方法 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);
}
}enqueue方法是EventBus#Poster接口中的，实现他的分别对应几个threadMode，这里按照顺序来分析一下 ASYNC MAIN MAIN_ORDERED BACKGROUND
ASYNC:
AsyncPoster#enqueue
public void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);//PendingPost 对象池
queue.enqueue(pendingPost);//queue 加入待处理事件队列
eventBus.getExecutorService().execute(this);//由于Async模式是每次都新开一个线程来执行所以直接添加进队里即可 通过newCachedThreadPool创建一个线程池 关于线程池不了解的可以单独去了解一下
}AsyncPoster#run
@Override
public void run() {
PendingPost pendingPost = queue.poll();//出队操作
if(pendingPost == null) {
throw new IllegalStateException("No pending post available");
}
eventBus.invokeSubscriber(pendingPost);
} void invokeSubscriber(PendingPost pendingPost) {
Object event = pendingPost.event;
Subscription subscription = pendingPost.subscription;
PendingPost.releasePendingPost(pendingPost);//初始化为null再加入缓存池中等待复用
if (subscription.active) {//active为false是调用了unregister 下面会分析到
invokeSubscriber(subscription, event);//这里就是调用了invokeSubscriber(subscription, event);和上面的一致了
}
}MAIN：
HandlerPoster#enqueue
public void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
synchronized (this) {
queue.enqueue(pendingPost);//将从对象池中取出的pendingPost放入待处理事件队列中
if (!handlerActive) {//只需要将事件不断加入到待处理队列即可 所以他是非阻塞式的
handlerActive = true;
if (!sendMessage(obtainMessage())) {
throw new EventBusException("Could not send handler message");
}
}
}
}
HandlerPoster#handleMessage
@Override
public void handleMessage(Message msg) {
boolean rescheduled = false;
try {
long started = SystemClock.uptimeMillis();
while (true) {
PendingPost pendingPost = queue.poll();
if (pendingPost == null) {
synchronized (this) {
// Check again, this time in synchronized
pendingPost = queue.poll();
if (pendingPost == null) {
handlerActive = false;
return;
}
}
}
eventBus.invokeSubscriber(pendingPost);//反射调用和上面的一致没啥再重复的了
long timeInMethod = SystemClock.uptimeMillis() - started;
if (timeInMethod >= maxMillisInsideHandleMessage) {
if (!sendMessage(obtainMessage())) {
throw new EventBusException("Could not send handler message");
}
rescheduled = true;
return;
}
}
} finally {
handlerActive = rescheduled;
}
}
MAIN_ORDERED:
和上面的MAIN一致走HANDLER
BACKGROUND:
BackgroundPoster#enqueue
public void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
synchronized (this) {
queue.enqueue(pendingPost);
if (!executorRunning) {//保证事件只能往待处理的列表中添加 而无法再开启线程 还记得开始的时候说BACKGROUND模式是共用一个后台进程吗
executorRunning = true;
eventBus.getExecutorService().execute(this);
}
}
}@Override
public void run() {
try {
try {
while (true) {
PendingPost pendingPost = queue.poll(1000);//如果这里没有获取到pendingPost会锁住1秒 1秒后调用poll
if (pendingPost == null) {
synchronized (this) {
// Check again, this time in synchronized
pendingPost = queue.poll();//双重检查 如果还没有待处理的事件添加进入待处理列表 结束线程
if (pendingPost == null) {
executorRunning = false;
return;
}
}
}
eventBus.invokeSubscriber(pendingPost);//这里和上面的一致了
}
} catch (InterruptedException e) {
eventBus.getLogger().log(Level.WARNING, Thread.currentThread().getName() + " was interruppted", e);
}
} finally {
executorRunning = false;
}
}post总结：
1.post调用后先使用ThreadLocal来存储事件,他可以隔离多个线程对数据的访问冲突
2.根据事件类型(也就是Post参数的类型)为key从subscriptionsByEventType获取订阅方法和订阅者,这个容器不了解的可以看下注册的总结
2.1此处我们已经具备了订阅者.订阅方法.待发送事件.post线程.threadmode等信息
3.根据线程和threadmode来判断
POSTING:直接反射调用(阻塞式调用)
MAIN:
发布在主线程:直接反射调用(阻塞式调用)
发布在子线程:通过Handler切换到主线程再调用反射(非阻塞式)
MAIN_ORDERED:默认走-通过Handler切换到主线程再调用反射(非阻塞式)
BACKGROUND:发布在主线程:通过线程池开启一条线程 事件将排队在同一条线程执行(非阻塞式)发布在子线程:直接反射调用(阻塞式调用)
ASYNC:总是开启一个新的线程来调用(非阻塞式)
ps:阻塞式:send:1 response:2 send:1 response:2 非阻塞式:send:1 send:1 response:2 response:2
这个例子如果看不明白那再解释一下,阻塞式-send1的时候阻塞了发布线程 直到订阅函数被执行后再send2
非阻塞式-将send1添加进入待处理事件列表之后 继续添加,不需要管订阅函数执行了没有
----------------------------------------------------------不华丽的风格陷---------------------------------------------
EventBus.getDefault().unregister(this);
EventBus#unregister
/** Unregisters the given subscriber from all event classes. */
public synchronized void unregister(Object subscriber) {
List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);//这个typesBySubscriber应该不陌生吧 忘了可以去注册的总结看一下 key-订阅者 value-list(事件类型)
if (subscribedTypes != null) {//遍历这个事件类型集合
for (Class<?> eventType : subscribedTypes) {
unsubscribeByEventType(subscriber, eventType);//删除subscriptionsByEventType中对应的订阅者和订阅方法 这样对订阅者的引用就不存在了,反之将持有订阅者的引用,导致泄露
}
typesBySubscriber.remove(subscriber);//删除订阅者对应的事件类型，这样对订阅者的引用就不存在了,反之将持有订阅者的引用,导致泄露
} else {
logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
}
}
EventBus#subscriptionsByEventType
/** Only updates subscriptionsByEventType, not typesBySubscriber! Caller must update typesBySubscriber. */
private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
//根据事件类型就可以从subscriptionsByEventType中获取到对应的订阅者和订阅方法 他们作为实体类Subscription存在List中
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);
//遍历List中所有的订阅者 如果有unregister相同的订阅者 则从List中删除
if (subscription.subscriber == subscriber) {
subscription.active = false;
subscriptions.remove(i);
i--; //防止角标越界,主要是因为List集合的特点是删除的不是lase就会自动向前进一位 举例 old:0 1 2 3 删除了0 new: 0 1 2
size--;//性能略好 减少for循环次数
}
}
}
}
unregister总结:
1.根据订阅者从typesBySubscriber中获取到所有的事件类型
2.有了事件类型列表就可以遍历事件类型从subscriptionsByEventType获取到对应的订阅者包括订阅函数来释放
3.根据订阅者删除掉typesBySubscriber中的事件类型 此时订阅者已经被释放不再持有该订阅者引用
----------------------------------------------------------不华丽的风格陷---------------------------------------------
再补上粘性事件:
EventBus.getDefault().postSticky(1);
EventBus#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);//走正常Post逻辑 上面已经分析过
}EventBus#subscribe
if (subscriberMethod.sticky) {//判断是否声明sticky为true
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>).
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);//postSticky时存储的粘性事件
}
}
} else {
Object stickyEvent = stickyEvents.get(eventType);
checkPostStickyEventToSubscription(newSubscription, stickyEvent);
}
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);
}
}又走到post的逻辑中来了,粘性事件结束.....
postSticky总结:其思想挺帅的,学到就是赚到,当然其应用场景也非常的多,尤其在不确定订阅者是否注册和已经销毁的情况下
----------------------------------------------------------不华丽的风格陷---------------------------------------------
好了,总算写完了,花了三天时间,第一天去看了一些博客再自己开始看源码,第二天和第三天开始写博客,碎片时间可能有笔误，或者有分析不好的地方,请指出
瓜分20万奖金 获得内推名额 丰厚实物奖励 易参与易上手
更多推荐
3
0- 0
已为社区贡献2条内容
所有评论(0)