PostThread
:默认的 ThreadMode
,表示在执行 Post
操作的线程直接调用订阅者的事件响应方法,不论该线程是否为主线程(UI 线程)。
当该线程为主线程时,响应方法中不能有耗时操作,否则有卡主线程的风险。
适用场景:对于是否在主线程执行无要求,但若 Post
线程为主线程,不能耗时的操作
MainThread
:在主线程中执行响应方法。如果发布线程就是主线程,则直接调用订阅者的事件响应方法,否则通过主线程的 Handler 发送消息在主线程中处理——调用订阅者的事件响应函数。显然,MainThread
类的方法也不能有耗时操作,以避免卡主线程。
适用场景:必须在主线程执行的操作
MAIN_ORDERED
:v3.1.1 中新增的属性,也是切换至主线程接收事件,并不会区分当前线程,直接走一遍handler的消息分发;
BackgroundThread
:在后台线程中执行响应方法。如果发布线程不是主线程,则直接调用订阅者的事件响应函数,否则启动唯一的后台线程去处理。由于后台线程是唯一的,当事件超过一个的时候,它们会被放在队列中依次执行,因此该类响应方法虽然没有PostThread
类和MainThread
类方法对性能敏感,但最好不要有重度耗时的操作或太频繁的轻度耗时操作,以造成其他操作等待。
适用场景:操作轻微耗时且不会过于频繁,即一般的耗时操作都可以放在这里
Async
:不论发布线程是否为主线程,都使用一个空闲线程来处理。它和BackgroundThread
不同的是,Async
类的所有线程是相互独立的,因此不会出现卡线程的问题。
适用场景:长耗时操作,例如网络访问。
首先我们需要将我们希望订阅事件的类,通过EventBus类注册,注册代码如下:
//3.0版本的注册
EventBus.getDefault().register(this);
//2.x版本的注册
EventBus.getDefault().register(this);
EventBus.getDefault().register(this, 100);
EventBus.getDefault().registerSticky(this, 100);
EventBus.getDefault().registerSticky(this);
可以看到2.x版本中有四种注册方法,区分了普通注册和粘性事件注册,并且在注册时可以选择接收事件的优先级,而3.0版本中的注册就变得简单,只有一个register()方法即可。
注册之后,我们需要编写响应事件的方法,代码如下:
//3.0版本
@Subscribe(threadMode = ThreadMode.BACKGROUND, sticky = true, priority = 100)
public void test(String str) {
}
//2.x版本
public void onEvent(String str) {
}
public void onEventMainThread(String str) {
}
public void onEventBackgroundThread(String str) {
}
在2.x版本中只有通过onEvent
开头的方法会被注册,而且响应事件方法触发的线程通过onEventMainThread
或onEventBackgroundThread
这些方法名区分,而在3.0版本中.通过@Subscribe
注解,来确定运行的线程threadMode
,是否接受粘性事件sticky
以及事件优先级priority
,而且方法名不在需要onEvent
开头,所以又简洁灵活了不少。
我们可以通过EventBus的post()
方法来发送事件,发送之后就会执行注册过这个事件的对应类的方法.或者通过postSticky()
来发送一个粘性事件.在代码是2.x版本和3.0版本是一样的.
EventBus.getDefault().post("str");
EventBus.getDefault().postSticky("str");
当不在需要接收事件的时候需要解除注册unregister
,2.x和3.0的解除注册也是相同的.代码如下:
EventBus.getDefault().unregister(this);
EventBus.getDefault()
使用了双重检查的单例模式创建保证线程安全
public static EventBus getDefault() {
if (defaultInstance == null) {
synchronized (EventBus.class) {
if (defaultInstance == null) {
defaultInstance = new EventBus();
}
}
}
return defaultInstance;
}
构造方法
使用了建造者模式,将复杂的对象与它的表示进行分离。实际上还用到了策略模式,其中Builder
中有些参数用于代码执行的策略,你传的参数不一样,我执行的方式不一样,像 ignoreGeneratedIndex
作用就是让 EventBus
如何查找出订阅方法的策略
EventBus(EventBusBuilder builder) {
//key 订阅事件,value所有订阅者的集合
//private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;
subscriptionsByEventType = new HashMap<>();
//key 订阅者 value 这个订阅者的事件集合
typesBySubscriber = new HashMap<>();
//粘性事件
stickyEvents = new ConcurrentHashMap<>();
//事件主线程处理
mainThreadPoster = new HandlerPoster(this, Looper.getMainLooper(), 10);
//事件Background处理
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;
}
EventBusBuilder
中使用了
private final static ExecutorService DEFAULT_EXECUTOR_SERVICE = Executors.newCachedThreadPool();
async
与background
线程模式都是使用这个线程池执行。
public void register(Object subscriber) {
//首先获取订阅者class对象
Class<?> subscriberClass = subscriber.getClass();
//通过subscriberMethodFinder查找订阅了那些事件,返回类型为SubscribeMethod方法,包含了订阅这个方法的method对象,
在那个线程订阅的threadmode,以及订阅的事件类型,以及是否是粘性事件。
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
//先从METHOD_CACHE查找是否有缓存,key:订阅类型,value:订阅方法名
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中
METHOD_CACHE.put(subscriberClass, subscriberMethods);
return subscriberMethods;
}
}
private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
//FindState 查找订阅方法的校验与保存
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
//通过反射获取订阅信息
findUsingReflectionInSingleClass(findState);
//调用父类方法用来校验订阅方法
findState.moveToSuperclass();
}
return getMethodsAndRelease(findState);
}
void moveToSuperclass() {
if (skipSuperClasses) {
clazz = null;
} else {
clazz = clazz.getSuperclass();
String clazzName = clazz.getName();
/** Skip system classes, this just degrades performance. */
//类名禁止以以下系统类名开头。
if (clazzName.startsWith("java.") || clazzName.startsWith("javax.") || clazzName.startsWith("android.")) {
clazz = null;
}
}
}
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();
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();
//实例化SubscribeMethod方法并添加
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");
}
}
}
subscribe
订阅
//必须在同步代码块中调用
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
//通过反射获取事件类型
Class<?> eventType = subscriberMethod.eventType;
//创建订阅
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
//判断是否已经添加过Subscription,添加过就抛异常
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();
//根据优先级添加subscriptions
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);
//判断是否是黏性事件,是的话立即发送
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);
}
}
}
public void post(Object event) {
//每一个线程都维护一个投递状态。currentPostingThreadState的实现是一个包含了PostingThreadState的ThreadLocal对象
//每个线程的数据都会互相不干扰
PostingThreadState postingState = currentPostingThreadState.get();
List<Object> eventQueue = postingState.eventQueue;
eventQueue.add(event);
if (!postingState.isPosting) {
//是否在主线程中进行消息循坏
postingState.isMainThread = Looper.getMainLooper() == Looper.myLooper();
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;
}
}
}
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);
//只要右边有一个为true,subscriptionFound就为true
subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
}
} else {
subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
}
//如果没发现
if (!subscriptionFound) {
if (logNoSubscriberMessages) {
Log.d(TAG, "No subscribers registered for event " + eventClass);
}
if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
eventClass != SubscriberExceptionEvent.class) {
post(new NoSubscriberEvent(this, event));
}
}
}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 {
//发送给订阅者,然后根据threadmode在不同线程中调用订阅者方法
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;
}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 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);
}
}
获取订阅者订阅类型,并移除订阅者,遍历集合取消订阅
public synchronized void unregister(Object subscriber) {
//通过typesBySubscriber获取这个订阅者的订阅类型
List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
if (subscribedTypes != null) {
for (Class<?> eventType : subscribedTypes) {
unsubscribeByEventType(subscriber, eventType);
}
//移除这个订阅者
typesBySubscriber.remove(subscriber);
} else {
Log.w(TAG, "Subscriber to unregister was not registered before: " + subscriber.getClass());
}
}
private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
//从subscriptionByEventType中获取这个事件类型的订阅者
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--;
}
}
}
}