简介
EventBus 是一款基于观察者模式实现的事件总线框架，由greenrobot开源。使用这个能简化app内的通信操作,能替代Intent, Handler, BroadCast 在 Fragment，Activity，Service，线程之间传递消息 。EventBus 将事件的接收者和发送者分开，可以通过较少的代码实现多个模块之间的通信，而不需要通过接口或者回调的方式来通信，能够降低通信带来的复杂度。分析源码的版本是3.1.1.
观察者模式
观察者模式（又被称为发布-订阅（Publish/Subscribe）模式，属于行为型模式的一种，它定义了一种一对多的依赖关系，让多个观察者对象同时监听某一个主题对象。这个主题对象在状态变化时，会通知所有的观察者对象，使他们能够自动更新自己。
在观察者模式中有如下角色：
Subject：抽象主题（抽象被观察者），抽象主题角色把所有观察者对象保存在一个集合里，每个主题都可以有任意数量的观察者，抽象主题提供一个接口，可以增加和删除观察者对象。
ConcreteSubject：具体主题（具体被观察者），该角色将有关状态存入具体观察者对象，在具体主题的内部状态发生改变时，给所有注册过的观察者发送通知。
Observer：抽象观察者，是观察者者的抽象类，它定义了一个更新接口，使得在得到主题更改通知时更新自己。
ConcrereObserver：具体观察者，实现抽象观察者定义的更新接口，以便在得到主题更改通知时更新自身的状态。
1577171354270.png
eventbus使用
eventbus的依赖比较简单，只需要在gradle文件中添加如下一行即可。不需要添加其他依赖，也较少出现和第三方sdk的依赖重复或冲突的问题。
implementation 'org.greenrobot:eventbus:3.1.1'
eventbus基于事件订阅的，因此就肯定有事件，订阅事件发送者，订阅事件接收者，取消订阅等模块。
首先需要定义不同的事件，以便用来区分事件的接收者和发送者。
public class EventBusParams {
public String key;
public Object object;
public EventBusParams(String key, Object object) {
this.key = key;
this.object = object;
}
}
定义了事件之后就可以发送，发送事件比较简单，只需要在通信发起的地方用一行代码就可以搞定。
EventBus.getDefault().post(new EventBusParams(key,object));
这样就能将key和object组成的event发送出去，有发送者就需要有接收者。任何地方都可以接收事件，只需要在对应的地方将自己注册成事件的接收者即可。注册的代码如下：
EventBus.getDefault().register(this);
注册一般是在事件发送之前就注册好，事件发送出去，订阅者能收到事件，但要处理事件还需要另外做一点工作。代码如下：
@Subscribe
public void handleEvent(EventBusParams event) {
if (event.key.equals(key)) {
// 如果是关注的事件key，就做相应的处理。
}
}
最后需要在合适的地方取消订阅，否则会发生内存泄漏，取消订阅的代码如下：
EventBus.getDefault().unregister(this);
从上面的流程看到，实际上eventbus的使用比较简单，不多的代码就能完成通信功能。为什么能做到高效通信，下面我们试图从源码分析的角度来解释。
源码分析
从下面这个图中可以看到，对于EventBus的工作过程很简单，用一句话概括为：事件被提交到EventBus后进行查找所有订阅该事件的方法并执行这些订阅方法。
1577175040578.png
接下来我们按照EventBus的使用流程来分析源码，即从eventbug的创建，注册事件，发送事件，接收事件的大体顺序来分析。
构造函数
在发送事件的时总是以EventBus.getDefault() 来获取实例，getDefault是单例方法，保证当前只有一个EventBus实例。相关方法如下：
private static final EventBusBuilder DEFAULT_BUILDER = new EventBusBuilder();
public static EventBus getDefault() {
if (defaultInstance == null) {
synchronized (EventBus.class) {
if (defaultInstance == null) {
defaultInstance = new EventBus();
}
}
}
return defaultInstance;
}
public EventBus() {
this(DEFAULT_BUILDER);
}
EventBus(EventBusBuilder builder) {
logger = builder.getLogger();
//以事件类型作为Key，Subscription的List集合作为Value的Map集合
subscriptionsByEventType = new HashMap<>();
//订阅者作为Key,订阅事件作为Value的Map集合
typesBySubscriber = new HashMap<>();
stickyEvents = new ConcurrentHashMap<>();
mainThreadSupport = builder.getMainThreadSupport();
//事件主线程处理
mainThreadPoster = mainThreadSupport != null ? mainThreadSupport.createPoster(this) : null;
backgroundPoster = new BackgroundPoster(this); //事件 Background 处理
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;
}
上面的DEFAULT_BUILDER是一个默认的EventBusBuilder，我们也可以通过配置EventBusBuilder来更改EventBus的属性。比如可以更改如下属性：
boolean logSubscriberExceptions = true;
boolean logNoSubscriberMessages = true;
boolean sendSubscriberExceptionEvent = true;
boolean sendNoSubscriberEvent = true;
boolean throwSubscriberException;
boolean eventInheritance = true;
注册
获取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和subscribe，下面先重点分析下findSubscriberMethods这个方法。subscriberMethodFinder这个对象是在eventbus 的构造函数中创建的，SubscriberMethodFinder的构造函数如下：
SubscriberMethodFinder(List<SubscriberInfoIndex> subscriberInfoIndexes, boolean strictMethodVerification,boolean ignoreGeneratedIndex) {
this.subscriberInfoIndexes = subscriberInfoIndexes;
this.strictMethodVerification = strictMethodVerification;
this.ignoreGeneratedIndex = ignoreGeneratedIndex;
}
subscriberInfoIndexes这个是用来处理eventbus的注解生成器相关的功能，注解生成器的详细介绍可以参考[老司机教你 “飙” EventBus 3]。 这里我们先跳过，可以认为subscriberInfoIndexes这个列表为空。strictMethodVerification和ignoreGeneratedIndex这两变量可以自己在定义EventBusBuilder时设置，默认值都为false。在这个基础上我们继续看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);
}
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;
}
}
static final Map<Class<?>, List<SubscriberMethod>> METHOD_CACHE = new ConcurrentHashMap<>();
METHOD_CACHE是一个ConcurrentHashMap，用来保存订阅类和订阅方法的对应关系，加快后续的查找速度。所以如果在METHOD_CACHE找到了subscriberClass多对应的subscriberMethods就直接返回，否则继续往下查找，如果最后查找到的列表为空，就会抛出一个EventBusException，这也就是说对于订阅者若是要成功注册到EventBus中，在订阅者中必须存在通过@Subscribe注解且为public类型的订阅方法。在成功获取到订阅方法集合以后，便将订阅方法集合添加到缓存中。对于这个缓存它是以订阅者的类作为key，订阅方法集合作为value的Map类型。我们重点看下中间的查找过程，前面讲到ignoreGeneratedIndex这个默认为false,所以我们直接看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();// 尝试查找父类的订阅方法
}
return getMethodsAndRelease(findState);
}
FindState是SubscriberMethodFinder的内部类，我们看下initForSubscriber这个方法，代码如下：
void initForSubscriber(Class<?> subscriberClass) {
this.subscriberClass = clazz = subscriberClass;
skipSuperClasses = false;
subscriberInfo = null;
}
在这个基础上接着分析findUsingInfo方法，因为不考虑注解，以及在initForSubscriber中将subscriberInfo置空，所以getSubscriberInfo这个方法也返回空。于是直接调用findUsingReflectionInSingleClass方法来查找订阅类的订阅方法，查找完成后会继续查找父类的订阅方法。查找结束后调用getMethodsAndRelease将查找到的订阅方法列表返回。下面重点分析下findUsingReflectionInSingleClass方法，代码如下：
private void findUsingReflectionInSingleClass(FindState findState) {
Method[] methods;
try {
// This is faster than getMethods
methods = findState.clazz.getDeclaredMethods();
} catch (Throwable th) {
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) {
//获取方法的Subscribe注解
Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
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()));
}
}
} 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");
}
}
}
从名字上看就知道这个是通过反射的形式获取订阅方法，主要流程是选择访问标志符是public，只有一个方法参数，并且被Subscribe注解标记的方法。随后将注解的信息和方法的信息组合创建一个SubscriberMethod实例，最后将实例加进findState的subscriberMethods列表中，至此就将订阅类的订阅方法通过键值对的形式关联起来。键是订阅类，值是该类中的订阅方法列表。前面讲到register有两个重要的方法，一个是上面分析的findSubscriberMethods，这个方法主要将订阅类和订阅方法关联起来。接下来我们分析另一个方法subscrib()，代码如下：
// Must be called in synchronized block
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
Class<?> eventType = subscriberMethod.eventType;
//根据订阅类和订阅方法构造一个订阅事件
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
//获取当前订阅事件中Subscription的List集合
CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions == null) {
//该事件对应的Subscription的List集合不存在，则重新创建并保存在subscriptionsByEventType中
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();
//将newSubscription按照订阅方法的优先级插入到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中
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);
}
}
}
在这个方法中便是订阅者真正的注册过程。首先会根据subscriber和subscriberMethod来创建一个Subscription对象。之后根据事件类型获取或创建一个Subscription集合subscriptions并添加到typesBySubscriber对象中。最后将刚才创建的Subscription对象添加到subscriptions之中。于是这样就完成了一次订阅者的注册操作。注册的主要流程如图所示:
1577262671100.png
发送事件
上面分析完注册之后，接下来看事件的发送。发送主要调用的是post()方法，代码如下：
public void post(Object event) {
//PostingThreadState保存着事件队列和线程状态信息
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;
}
}
}
首先是通过currentPostingThreadState.get()方法来得到当前线程PostingThreadState的对象,为什么是说当前线程我们来看看currentPostingThreadState的实现:
private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {
@Override
protected PostingThreadState initialValue() {
return new PostingThreadState();
}
};
currentPostingThreadState的实现是一个包含了PostingThreadState对象的ThreadLocal对象，ThreadLocal是一个线程内部的数据存储类。通过它可以在指定的线程存储数据，而这个数据是线程私有的，其他线程访问不到的。内部原理是通过生成一个泛型数据的数组，在不同的线程会有不同的数组索引，所以通过get获取数据时，取到的就是当前线程对应的数据。ThreadLocal 的详细介绍可以参考[Java多线程基础-ThreadLocal]。post方法首先从PostingThreadState对象中取出事件队列，然后再将当前的事件插入到事件队列当中。最后将队列中的事件依次交由postSingleEvent方法进行处理，并移除该事件。这个方法最终将事件发送给响应的订阅者，我们重点看下这个方法，代码如下：
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
//eventInheritance表示是否向上查找事件的父类,默认为true
if (eventInheritance) {
//当eventInheritance为true时，则通过lookupAllEventTypes找到所有的父类事件并存发在List中
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);
}
if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
eventClass != SubscriberExceptionEvent.class) {
post(new NoSubscriberEvent(this, event));
}
}
}
先通过lookupAllEventTypes找到所有的父类事件并存发在List中，然后通过postSingleEventForEventType对事件逐一处理，下面接着看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;
}
subscriptionsByEventType是一个hashmap,其中的键值对在上面的register中添加。键为订阅类，值是保存订阅类和订阅方法的Subscription组成的列表。对于订阅类所对应的Subscription列表，对每个Subscription调用postToSubscription方法。代码如下：
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);
}
}
我们在接收事件的方法的注解上可以定义处理的线程模式，默认的模式是POSTING。在这里取出订阅方法的线程模式，之后根据订阅方法所设置的线程模式来选择线程来执行订阅方法的线程。 如果是默认的POSTING，就会直接调用invokeSubscriber来讲消息发送出去，在那个线程发送事件就在那个线程处理事件。如果是其他的类型，要么通过相应的poster将事件入队，要么调用invokeSubscriber方法。下面先看下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);
}
}
这个是通过反射的方式来执行订阅事件的方法，这样事件就发送出去并被订阅者接收并做相应处理了。下面来看通过poster的方式处理事件，poster可以理解为将事件传递到特定线程执行的操作类。下面以HandlerPoster 为例简单分析下，代码如下：
public class HandlerPoster extends Handler implements Poster {
private final PendingPostQueue queue;
private final int maxMillisInsideHandleMessage;
private final EventBus eventBus;
private boolean handlerActive;
protected HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
super(looper);
this.eventBus = eventBus;
this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
queue = new PendingPostQueue();
}
public void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
synchronized (this) {
queue.enqueue(pendingPost);// 入队列
if (!handlerActive) {
handlerActive = true;
// 发送开始处理事件的消息，handleMessage()方法将被执行，完成从子线程到主线程的切换
if (!sendMessage(obtainMessage())) {
throw new EventBusException("Could not send handler message");
}
}
}
}
@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);// 进一步处理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;
}
}
}
HandlerPoster的enqueue()方法主要就是将subscription、event对象封装成一个PendingPost对象，然后保存到队列里，之后通过Handler切换到主线程，在handleMessage()方法将中将PendingPost对象循环出队列，交给invokeSubscriber()方法做进一步处理。
void invokeSubscriber(PendingPost pendingPost) {
Object event = pendingPost.event;
Subscription subscription = pendingPost.subscription;
PendingPost.releasePendingPost(pendingPost);
if (subscription.active) {
invokeSubscriber(subscription, event);
}
}
invokeSubscriber方法比较简单，主要就是从pendingPost中取出之前保存的event、subscription，然后用反射来执行订阅事件的方法，又回到了第一种处理方式。所以mainThreadPoster.enqueue(subscription, event)的核心就是先将将事件入队列，然后通过Handler从子线程切换到主线程中去处理事件。
backgroundPoster.enqueue()和asyncPoster.enqueue也类似，内部都是先将事件入队列，然后再出队列，但是会通过线程池去进一步处理事件。 至此，eventbus就通过post方法之后将消息传递给了订阅类的订阅方法。
取消注册
前面分析了注册以及事件发送的流程，下面分析取消注册的过程。取消注册只需要调用unregister方法就可，代码如下：
public synchronized void unregister(Object subscriber) {
//通过typesBySubscriber来取出这个subscriber订阅者订阅的事件类型,
List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
if (subscribedTypes != null) {
for (Class<?> eventType : subscribedTypes) {
unsubscribeByEventType(subscriber, eventType);
}
//从typesBySubscriber移除subscriber
typesBySubscriber.remove(subscriber);
} else {
logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
}
}
取消注册也是针对订阅类来的，针对subscriber这个订阅者取出对应的订阅事件类型。最终调用unsubscribeByEventType取消事件和订阅方法的关联，代码如下：
private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
// 得到当前参数类型对应的Subscription集合
List<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions != null) {
// 遍历Subscription集合
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--;
}
}
}
}
最终分别从typesBySubscriber和subscriptions里分别移除订阅者以及相关信息即可。在unregister()方法中，最主要的就是释放typesBySubscriber、subscriptionsByEventType中缓存的资源。
粘性事件
前面我们都是先注册后，然后在订阅方法中等待事件的到来。但黏性事件的存在改变了先注册接收再发送这一顺序，这个功能比较惊艳。下面一鼓作气也分析下粘性事件的发送接收过程。
粘性事件的发送有点特殊，调用的方法是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);
}
postSticky方法主要做了两件事，首先将事件类型和事件保存在stickyEvents中，然后将事件通过post方法发送出去，这个和普通事件的传递一样。所以在发送黏性事件的时候，已经订阅了该类型的订阅方法还是一样可以收到这个事件。对于在发送之后再注册的订阅者，后续还能收到的原因就在第一步的stickyEvents中。粘性事件的特色就是先发送事件再进行注册，注册就会用到之前register方法，这个方法还是会调用subscribe() ，这个方法之前之前分析的时候跳过了黏性事件的处理。下面我们分析这一部分，代码如下：
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
... // 省略前面已分析代码
if (subscriberMethod.sticky) {// 下面是粘性事件的处理
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);
}
}
} else {
Object stickyEvent = stickyEvents.get(eventType);
checkPostStickyEventToSubscription(newSubscription, stickyEvent);
}
}
}
处理粘性事件的关键在于eventbus注册时，遍历stickyEvents 。如果当前要注册的事件订阅方法是粘性的，并且该方法接收的事件类型和stickyEvents 中某得事件的类型相同或存在父子关系，则取出stickyEvents 中对应事件的具体类型，走进一步处理。对于粘性事件，在注册阶段要关注的方法是checkPostStickyEventToSubscription，这个将已经发出的事件重新发给新注册的方法。代码如下：
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, isMainThread());
}
}
而postToSubscription这个方法我们前面已经分析过了，只不过是在普通事件的发送过程。因此粘性事件后面注册的方法能接受到先发送的事件，就在于后面注册的方法在注册阶段就会遍历前面已经发送的黏性事件集合，从中找出关注的事件，最后再将事件重新发送一次。普通事件就是先注册，后面发送的时候就直接找订阅的方法。而黏性事件简单说就是会发送两次，一次是发送的时候将事件发送给已经注册的方法，一次是后面注册的方法在注册阶段会对新注册的方法再发送一次。这么看来，黏性事件看上去很神奇，但实际并不算太复杂。
总结
值得借鉴的地方
单例模式对外提供统一的实例，
提供可配置项，但默认使用建造者模式创建，方便使用也提供了自定义的高阶选项。
使用CopyOnWriteArrayList，虽然是单例对外提供，但也做好了线程同步等措施。
使用threadlocal保存线程独有数据。
提供注解满足各种使用场景的要求，比如线程，优先级等。
使用几种poster满足不同的线程模型。
另外对于开源框架可以加入日志断点，能显著帮助代码流程的梳理。
EventBus还有个重要的功能点，索引功能这里没有分析。感兴趣的可以参考[老司机教你 “飙” EventBus 3]
eventbus从使用上非常方便，不需要自己实现通信。只需要在发送事件的地方发送，在接收的地方注册并监听，就能很随意获取到任何回调的结果，也不需要管理线程的切换。模块之间的交互更加简单。但eventbus也存在些许缺陷，比如通信只能单向，事件过多会导致系统混乱，改动起来容易遗漏。
介绍eventbus的文章很多，这里写下就权当给自己的备忘。
感谢先行者的代码开源和分析文章，有错的地方请指出，感谢。
网友评论