Eventbus 使用方法和原理分析

3、添加订阅方法

回想观察者模式，观察者有着一个 update() 方法，在接收到事件的时候会调用该 update() 方法，这个方法就是一个订阅方法。在EventBus 3.0中，声明一个订阅方法需要用到 @Subscribe 注解，因此在订阅者类中添加一个有着 @Subscribe 注解的方法即可，方法名字可自定义，而且必须是public权限，其方法参数有且只能有一个，另外类型必须为第一步定义好的事件类型(比如上面的 MessageEvent)，如下所示：

完整的 MainActivity.java 文件如下所示：

4、发送事件

与观察者模式对应的，当有事件发生，需要通知观察者的时候，被观察者会调用 notifyObservers() 方法来通知所有已经注册的观察者，在 EventBus 中，对观察者模式底层进行了封装，我们只需要调用以下代码就能把事件发送出去：

上述 EventType 就是第一步定义的事件类型。

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 :

Android EventBus3.1.1从使用到源码解析

/** 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 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);//分类保存后面有分析
}
}
}

// 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);
}
}
}

HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
super(looper);
this.eventBus = eventBus;
this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
queue = new PendingPostQueue();//采用独立队列，与backgroundPoster一致
}

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);
}
}

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;
}
}

public void post(Object event) {
//currentPostingThreadState是ThreadLocal,ThreadLocal可以解决多线程的并发访问问题,他会为每一个线程提供一个独立的变量副本,可以隔离多个线程对数据的访问冲突
PostingThreadState postingState = currentPostingThreadState.get();
......
}

final static class PostingThreadState {
// 通过post方法参数传入的事件集合
final List<Object> eventQueue = new ArrayList<Object>();
boolean isPosting; // 是否正在执行postSingleEvent()方法
boolean isMainThread;
Subscription subscription;
Object event;
boolean canceled;
}

Subscription(Object subscriber, SubscriberMethod subscriberMethod) {
this.subscriber = subscriber;
this.subscriberMethod = subscriberMethod;
active = true;
}

public SubscriberMethod(String methodName, Class<?> eventType, ThreadMode threadMode,
int priority, boolean sticky) {
this.methodName = methodName;
this.threadMode = threadMode;
this.eventType = eventType;
this.priority = priority;
this.sticky = sticky;
}

synchronized (this) {
//这里根据我们注册的时候总结 这个容器中装的是订阅者和订阅方法,现在根据发送事件的类型来获取到对应的订阅者和订阅方法这些参数是反射必须要用到的
subscriptions = subscriptionsByEventType.get(eventClass);
}

//这里根据是否在主线程和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);
}
}

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
//获取subsrciberMethod传递的自定义EventType参数的运行时的类
Class eventType = subscriberMethod.eventType;
//Subscription用于绑定subscriber和sucriberMethod,一个订阅者可以有多个subscriberMethod
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
//根据EventType的运行时类取到该类所有的subscriptioins，subscriptionsByEventType是HashMap中的key
CopyOnWriteArrayList subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions == null) {
subscriptions = new CopyOnWriteArrayList<>();
//若根据EventType找不到subscriptions,则eventType作key,subscriptions作value添加到subscriptionByEventType中。
subscriptionsByEventType.put(eventType, subscriptions);
} else {
if (subscriptions.contains(newSubscription)) {
//已经存在newSubscription，抛出异常该订阅者已经注册，不可重复注册同一个subscriber
throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
+ eventType);
}
} int size = subscriptions.size();
for (int i = 0; i <= size; i++) {
//循环subscriptions，根据标记优先级的priority从高到低，将新的subscription插入到subscriptions中
if (i == size || subscriberMethod.priority > subscriptions.get(i).subscriberMethod.priority) {
subscriptions.add(i, newSubscription);
break;
}
}
//typesBySubscriber是一个HashMap，根据subscriber做key,获取该subscriber对应的所有的订阅事件的类型
List> subscribedEvents = typesBySubscriber.get(subscriber);
if (subscribedEvents == null) {
subscribedEvents = new ArrayList<>();
//该订阅者之前的订阅事件类型列表为空，则将当前订阅类型添加到typesBySubscriber中
typesBySubscriber.put(subscriber, subscribedEvents);
}
subscribedEvents.add(eventType);
//如果该方法被标识为sticky事件
if (subscriberMethod.sticky) {
if (eventInheritance) { eventInheritance标识是否考虑EventType的类层次结构
//循环所有的sticky黏性事件
Set, Object>> entries = stickyEvents.entrySet();
for (Map.Entry, Object> entry : entries) {
Class candidateEventType = entry.getKey();
//如果当前事件是其他事件的同类型的或者是他们的父类
if (eventType.isAssignableFrom(candidateEventType)) {
Object stickyEvent = entry.getValue();
heckPostStickyEventToSubscription(newSubscription, stickyEvent);
}
}
} else {
Object stickyEvent = stickyEvents.get(eventType);
checkPostStickyEventToSubscription(newSubscription, stickyEvent);
}
}
}

private void checkPostStickyEventToSubscription(Subscription newSubscription, Object stickyEvent) {
if (stickyEvent != null) {
// If the subscriber is trying to abort the event, it will fail (event is not tracked in posting state)
// --> Strange corner case, which we don't take care of here.
postToSubscription(newSubscription, stickyEvent, Looper.getMainLooper() == Looper.myLooper());
}
}

private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
//根据@subscriber中threadMode进行区分，POSTING为当前线程执行，
//MAIN为主线程，BACKGROUND为子进程，ASYNC为异步执行。
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);
}
}