EventBus3.0 源码解析

基本使用

EventBus是专门为Android设计的用于订阅，发布总线的库，用到这个库的app很多，因为它有很多的优点。比如，它可以简单android组件之间的通信；它可以避免了android四大组件复杂的生命周期处理；它可以让你的代码更为简洁。先一起了解下如何使用，然后在分析它的源码，知道它的工作原理。我们直接来使用EventBus 3.0,3.x主要的一个新的特性就是使用了注解，我们的Subscribe可以在代码中就指定我们的EventBus使用什么ThreadMode，是否粘性事件，优先级。

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

这些，等会在源码分析的时候会讲，先看看如何使用：

public class MainActivity extends AppCompatActivity {

private TextView tv;
private Button btn;
private NetTask netTask;

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);

EventBus.getDefault().register(this);
netTask = new NetTask();

tv = (TextView)findViewById(R.id.first_tv);
btn = (Button)findViewById(R.id.first_btn);
btn.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
netTask.runTask();
}
});
}

@Subscribe(threadMode = ThreadMode.MAIN)
public void onEventMainThread(Object object) {
if(object instanceof NetTask){
String result= ((NetTask) object).getResult();
tv.setText(result);
}
}

@Override
protected void onDestroy() {
try{

EventBus.getDefault().unregister(this);

}catch (Exception e){
}
super.onDestroy();
}
}

视图上显示一个TextView和一个Button，点击按钮的时候，就执行NetWork任务请求，请求执行结束后，在onEventMainThread方法更新UI，下面是NetWork :

public class NetTask {

private String result = "";

private static final int SUCCESS = 0x23;
private static final int ERROR = 0x24;

public String getResult(){
return result;
}

public void runTask(){

new AsyncTask<Void,Void,Integer>(){

@Override
protected Integer doInBackground(Void... params) {

//模拟网络请求
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}

result = "Event Bus";
return SUCCESS;
}

@Override
protected void onPostExecute(Integer integer) {
if(integer == SUCCESS){
EventBus.getDefault().post(NetTask.this);
}
}
}.execute();
}

}

代码看起来要比广播和回调方便多了，下面是UI显示：

源码分析

初始化

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) {
subscriptionsByEventType = new HashMap<>();
typesBySubscriber = new HashMap<>();
stickyEvents = new ConcurrentHashMap<>();
mainThreadPoster = new HandlerPoster(this, Looper.getMainLooper(), 10);
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;
}

上面是EventBus初始化的三个步骤，直观上看用到单例模式和Builder模式，将构造参数给分离了出来，实际上还用到了策略模式，其中Builder中有些参数用于代码执行的策略，就说，你传的参数不一样，我执行的方式不一样，像ignoreGeneratedIndex 作用就是让EventBus如何查找出订阅方法的策略。这些布尔类型的参数，在分析代码中可以逐步的了解到，先了解一些缓存对象，以更容易的了解源码：

subscriptionsByEventType : 内部是一个Map集合，可以根据EventType查找订阅事件

typesBySubscriber : 根据我们的订阅对象找到EventType

stickyEvents : 粘性事件的缓存

事件投递者 : mainThreadPoster,backgroundPoster,asyncPoster 根据订阅注解ThreadMode去选择不同的投递者，不同投递者投递事件，接收函数会执行在不同的线程中

subscriberMethodFinder ：查找方法用的，内部维护了一个订阅方法的集合

注册

public void register(Object subscriber) {
Class<?> subscriberClass = subscriber.getClass();
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}

当我们调用register(this)的时候就把订阅者给传了进来，代码量很少，主要就两个步骤，第一个findSubscriberMethods找出一个SubscriberMethod的集合，然后就遍历SubscriberMethod去订阅事件，我们先看看findSubscriberMethods()里面到底做了什么，返回的是什么。

List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {

List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}

/**
* ignoreGeneratedIndex 为true则表明用反射生成 subscriberMethods
* false则使用apt获取 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.put(subscriberClass, subscriberMethods);
return subscriberMethods;
}
}
}

首先从缓存中查找，如果找到了就立马返回。如果缓存中没有的话，则根据 ignoreGeneratedIndex 选择如何查找订阅方法，最后，找到订阅方法后，放入缓存，以免下次继续查找。ignoreGeneratedIndex 默认就是false，那我们会执行findUsingInfo()方法，但是这里先分析findUsingReflection()，因为在我们没有做任何配置的情况下还是会执行上面的findUsingReflection()，就是通过反射来解析注解。

private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
findUsingReflectionInSingleClass(findState);
findState.moveToSuperclass();
}
return getMethodsAndRelease(findState);
}

从上面的代码块，我们可以看到，第一步准备FindState，我们点进去看看：

private FindState prepareFindState() {
synchronized (FIND_STATE_POOL) {
for (int i = 0; i < POOL_SIZE; i++) {
FindState state = FIND_STATE_POOL[i];
if (state != null) {
FIND_STATE_POOL[i] = null;
return state;
}
}
}
return new FindState();
}

这里我们先了解从池中拿出一个FindState对象，FindState中维护的就是我们对订阅方法查找结果的封装。其实，往后面，我们就知道作者这里设计的非常精妙。第二步，initForSubscriber()就是将我们的订阅者传给FindState对象。第三步做的就是不断从订阅者和订阅者的父类去查找订阅方法，一起看findUsingReflectionInSingleClass()。

/**
* 用发射对注解的处理
* @param findState
*/
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();
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");
}
}
}

代码很长，其实就是对订阅者方法的遍历，看有没有注解，有的话就解析注解，最后将找到的订阅方法的集合封装到FindState对象中的subscriberMethods集合中。我们解析完了之后，在看findUsingReflection()方法的最后，返回了getMethodsAndRelease(FindState)，将我们的FindState穿给了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;
break;
}
}
}
return subscriberMethods;
}

从这里，我们就知道作者设计FindState池的初心了，解析完了之后，将订阅方法赋给List集合，再回收FindState，继续接收解析，内存没有半点浪费。最后，我们返回的是一个订阅方法的集合。这样，我们通过反射解析注解，找到订阅方法的方式，我们已经分析完了。再看看通过apt处理器来找，我们知道apt处理，是针对源码的处理，是执行在编译过程中的。所以性能要比反射好的多，所以推荐大家使用这种方式。好，那最开始，我说使用findUsingInfo()方法如果没有配置还是使用反射呢，我们一起看看：

private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
/**
* prepareFindState从池中获取FindState
*/
FindState findState = prepareFindState();
/**
* 把 订阅类交给 FindState
*/
findState.initForSubscriber(subscriberClass);
//什么时候为false，就是当当前类和父类的订阅方法全部检查完
while (findState.clazz != null) {
//从EventBusIndex中查找订阅信息
/**
* SubscriberInfo中维护的对象
* private final Class subscriberClass;
* private final Class<? extends SubscriberInfo> superSubscriberInfoClass;
* private final boolean shouldCheckSuperclass;
* private final SubscriberMethodInfo[] methodInfos;
*/
findState.subscriberInfo = getSubscriberInfo(findState);
if (findState.subscriberInfo != null) {
SubscriberMethod[] array = findState.subscriberInfo.getSubscriberMethods();
//根据订阅信息遍历他的订阅方法
/**
* SubscriberMethod
*final String methodName;
*final ThreadMode threadMode;
*final Class<?> eventType;
*final int priority;
*final boolean sticky;
*/
for (SubscriberMethod subscriberMethod : array) {
/*
* 根据eventType对SubscriberMethod进行检查
* 如果有这种类型的方法，或者有这个方法的类型的子类就返回false了
*/
if (findState.checkAdd(subscriberMethod.method, subscriberMethod.eventType)) {
findState.subscriberMethods.add(subscriberMethod);
}
}
} else {
//如果没有EventBusIndex的信息，则用反射处理注解
findUsingReflectionInSingleClass(findState);
}
//移动到父类
findState.moveToSuperclass();
}
//置空FindState池
return getMethodsAndRelease(findState);
}

上面，我注释解析的很清楚了，前几步的操作基本一致，主要看getSubscriberInfo(findState)这个方法：

private SubscriberInfo getSubscriberInfo(FindState findState) {
//查找是否存在订阅信息，有则直接返回
if (findState.subscriberInfo != null && findState.subscriberInfo.getSuperSubscriberInfo() != null) {
SubscriberInfo superclassInfo = findState.subscriberInfo.getSuperSubscriberInfo();
if (findState.clazz == superclassInfo.getSubscriberClass()) {
return superclassInfo;
}
}
//从我们传入的MyEventBusIndex中查找是否存在订阅信息，订阅信息就是订阅方法，包括父类的订阅方法
if (subscriberInfoIndexes != null) {
for (SubscriberInfoIndex index : subscriberInfoIndexes) {
SubscriberInfo info = index.getSubscriberInfo(findState.clazz);
if (info != null) {
return info;
}
}
}
return null;
}

上面一段代码是针对缓存的，如果缓存中有就直接返回，底下，是从subscriberInfoIndexes中查找订阅信息的。那么，我们的subscriberInfoIndexes从哪来的呢，是EventBus Builder对象，

public class EventBusBuilder {
private final static ExecutorService DEFAULT_EXECUTOR_SERVICE = Executors.newCachedThreadPool();

......

List<SubscriberInfoIndex> subscriberInfoIndexes;

默认是空的，所以结合上面的代码，还是执行了findUsingReflection()方法。那么，我们应该如何操作呢？我们需要配置下。第一步引入EventBusAnnotationProcessor库



这个库是在预编译过程，对我们的代码进行处理，找到我们有订阅方法的类的。第二步，配置项目的build.gradle文件。



将我们的库和andorid-apt引入，其中参数arguments是指定，我们处理完我们的源码文件，生成的文件名的。好，我们rebuild一下我们的项目。然后在build文件夹下面发现了：



我们点进去看看，



把我们的订阅类和订阅方法全找出来了，当，我们这个类初始化的时候，我们的SUBSCRIBER_INDEX这个集合就已经有我们的订阅信息了。所以，你知道为什么，叫ignoreGeneratedIndex了把。好，最后一步，我们就通过EventBus初始化，将这个类给传进去。

EventBus.builder().addIndex(new MyEventBusIndex()).
installDefaultEventBus().register(this);

这样，它就会通过apt方式来处理我们的注解了，需要强调的是，我们是在编译过程中就已经找到我们的订阅信息了，不会在运行期做任何事情，所以这个效率是非常高的。好了，我们查找订阅的代码已经全部分析完了，这个部分是3.0最核心的代码，作者很精心的设计以提高EventBus的性能，我们看完之后也受益良多啊。好，继续分析，继续看findUsingInfo()方法，之前，我注释的也很清楚，但是还有个地方需要注意：

for (SubscriberMethod subscriberMethod : array) {
/*
* 根据eventType对SubscriberMethod进行检查
* 如果有这种类型的方法，或者有这个方法的类型的子类就返回false了
*/
if (findState.checkAdd(subscriberMethod.method, subscriberMethod.eventType)) {
findState.subscriberMethods.add(subscriberMethod);
}
}

我们添加订阅方法的时候还有个检查，我们分析下：

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.
//两层检查：一层是只根据EventType，第二层根据签名
Object existing = anyMethodByEventType.put(eventType, method);
if (existing == null) {
return true;
} else {
if (existing instanceof Method) {
//只有子类中没有发现这种类型的方法类型才返回true
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);
}
}

这里，有两步检查，第一步检查，很好懂，第二步，根据什么签名来检查，我们看看：

private boolean checkAddWithMethodSignature(Method method, Class<?> eventType) {
methodKeyBuilder.setLength(0);
methodKeyBuilder.append(method.getName());
methodKeyBuilder.append('>').append(eventType.getName());

//将方法名和事件类型当作key，保存方法
String methodKey = methodKeyBuilder.toString();
Class<?> methodClass = method.getDeclaringClass();
Class<?> methodClassOld = subscriberClassByMethodKey.put(methodKey, methodClass);

//如果我们传递过来的methodClass是父类，则直接返回
if (methodClassOld == null || methodClassOld.isAssignableFrom(methodClass)) {
// Only add if not already found in a sub class
return true;
} else {

//否则，保存我们的签名的MethodClass
// Revert the put, old class is further down the class hierarchy
subscriberClassByMethodKey.put(methodKey, methodClassOld);
return false;
}
}

其实作者，这里又做了一个优化，将方法名和事件类型当作key，来保存方法，将传来的方法类型和我们签名的保存的比较，如果我们保存的是父类，就返回true，如果是子类，就将传来的方法保存起来，返回false。这样做的意图是，如果有父类的方法了，就没有必要添加子类的方法了，因为继承会执行到的。至此，我们对查找订阅方法的过程已经完全分析完了。看懂了之后，非常的过瘾。无论哪种方式查找，都返回了SubscriberMethod对象，我们看看它维护了什么属性：

/** Used internally by EventBus and generated subscriber indexes. */
public class SubscriberMethod {
final Method method;
final ThreadMode threadMode;
final Class<?> eventType;
final int priority;
final boolean sticky;
/** Used for efficient comparison */
String methodString;

可以看到，基本都是我们注解标识的内容。好，我们继续分析订阅过程：

2.订阅

// Must be called in synchronized block
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
Class<?> eventType = subscriberMethod.eventType;

/**
* Subsciption
* final Object subscriber;
* final SubscriberMethod subscriberMethod;
* volatile boolean active; 如果为false，那么当unregister(this)调用的时候，那么在队列中检查，以防止竞争条件
*/
//根据订阅者和订阅方法构造一个订阅事件
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);

//subscriptionsByEventType 是根据eventType去查找Subsciption的集合
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);
}
}

/**
* 遍历订阅事件，找到比subscriptions中订阅事件小的位置，然后插进去
* 这样就可以根据优先级依次投递事件了
*/
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是根据订阅者去查找EventType的缓存
//为了是unregister(this)，根据this,去解绑事件的。
List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
if (subscribedEvents == null) {
subscribedEvents = new ArrayList<>();
typesBySubscriber.put(subscriber, subscribedEvents);
}
subscribedEvents.add(eventType);

//如果是粘性事件，就立马处理
if (subscriberMethod.sticky) {
if (eventInheritance) {

//EventType的子类也应该考虑

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

订阅的代码比较长，但是理解起来并不难，而且，我也做了很详细的注释，其实里面就做了两件事，将我们的订阅方法和订阅者，封装到subscriptionsByEventType和typesBySubscriber，至于这两个对象是干什么的呢？第一个是我们投递订阅事件的时候，就是根据我们的EventType找到我们的订阅事件，从而去分发事件，处理事件的；第二个是在调用unregister(this)的时候，根据订阅者找到我们的EventType，又根据我们的EventType找到订阅事件，从而解绑用的，由于篇幅问题就不详细分析了。第二件事，就是如果是粘性事件的话，就立马投递、执行。

3.发布

/** Posts the given event to the event bus. */
public void post(Object event) {
/**
* PostingThreadState:
* final List<Object> eventQueue = new ArrayList<Object>();
* boolean isPosting;
* boolean isMainThread;
* Subscription subscription;
* Object event;
* boolean canceled;
*/
//每个线程中都维护了一个投递的状态
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;
}
}
}

当我们调用post(Object object)的方法的时候就执行了上面的代码，PostingThreadState是维护了投递的状态，最后循环投递，直到PostingThreadState中的EventQueue为空。那么代码最终执行到：

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

那么，我们就会根据ThreadMode去处理事件了。也是由于篇幅的问题，就分析一种了，当线程模式是主线程的时候，意味着，我们需要执行的代码在主线程中操作。如果是主线程，就是通过反射，直接运行订阅的方法，如果不是主线程，我们需要mainThreadPoster将我们的订阅事件入队列，一起看看mainThreadPoster的工作原理：

final class HandlerPoster extends Handler {

private final PendingPostQueue queue;
private final int maxMillisInsideHandleMessage;
private final EventBus eventBus;
private boolean handlerActive;

HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
super(looper);
this.eventBus = eventBus;
this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
queue = new PendingPostQueue();
}

void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
synchronized (this) {
queue.enqueue(pendingPost);
if (!handlerActive) {
handlerActive = true;
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);
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;
}
}
}

其实，在EventBus初始化的时候，mainThreadPoster就已经获取主线程的Looper了，就是用到了我们Android的消息处理机制，Looper，Handler。至于消息队列是自己维护的一个单向的链表。每次向Andorid的主线程Looper投递一个空消息，然后在HandlerMessage()方法里面从自己维护的队列中取出PendingPost 进行处理。

final class PendingPost {
private final static List<PendingPost> pendingPostPool = new ArrayList<PendingPost>();

Object event;
Subscription subscription;
PendingPost next;

而PendingPost中维护的是订阅事件，EventType和下一个PendingPost的地址。

4.反注册

public synchronized void unregister(Object subscriber) {
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());
}
}

反注册就是通过，我们EventBus中typesBySubscriber这个属性，通过订阅者去查找订阅事件，然后去一一解绑的。当然，反注册主要是为了提高效率的，不然订阅的事件太多，非常影响性能。

5.新特性-粘性事件

是否还记得，我们看订阅中的最后一段代码呢？不记得，也没事，继续看：

//如果是粘性事件，就立马处理
if (subscriberMethod.sticky) {
if (eventInheritance) {

//EventType的子类也应该考虑

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

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

看到没有，如果我们在注册的时候，指定我们要发布粘性事件，那么我们在订阅的时候，就立马调用postToSubscription，去发布了，至于它从缓存中stickyEvents中获取订阅事件，你可能要问，我们什么时候把EventType放进去的呢？

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()方法的时候，放入了缓存。当然，你也可以移除粘性事件。

public boolean removeStickyEvent(Object event) {
synchronized (stickyEvents) {
Class<?> eventType = event.getClass();
Object existingEvent = stickyEvents.get(eventType);
if (event.equals(existingEvent)) {
stickyEvents.remove(eventType);
return true;
} else {
return false;
}
}
}

根据源码的分析，我们把上面的代码改成粘性事件。首先，在NetWork里面：

@Override
protected void onPostExecute(Integer integer) {
if(integer == SUCCESS){
EventBus.getDefault().postSticky(NetTask.this);
}
}

然后，在MainActivity跳转到SecondActivity，让SecondActivity粘性注册：

public class SecondActivity extends AppCompatActivity {

private TextView tv2;
private Handler handler = new Handler();

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_second);

EventBus.getDefault().registerSticky(this);

tv2 = (TextView) findViewById(R.id.second_tv);
}

@Subscribe(threadMode = ThreadMode.MAIN,sticky = true)
public void onEventMainThread(final NetTask netTask) {
handler.post(new Runnable() {
@Override
public void run() {
String result= netTask.getResult();
tv2.setText(result);
}
});
}
}

这样，我们只要一打开SecondActivity，TextView上面就会显示Event Bus，下面是UI显示：

思考

分析了好久的源码，我们收获很多，除了对框架的理解得到了升华，其次，我们也体会到了作者的设计思想，这在我们平常工作中都可以去尝试，体验 。最后，我们来思考下使用它的优点。首先和Android自带的组件广播相比，除了代码简洁之外，更多的是，我们能获取到更多的信息，众所周知，观察者有两种模式，一种被观察者推，一种是观察者自己拉取。

@Override
protected void onPostExecute(Integer integer) {
if(integer == SUCCESS){
EventBus.getDefault().postSticky(NetTask.this);
}
}

@Subscribe(threadMode = ThreadMode.MAIN,sticky = true)
public void onEventMainThread(Object object) {
if(object instanceof NetTask){
String result= ((NetTask) object).getResult();
tv.setText(result);
}
}

我们可以在投递的过程把被观察着本身投递出去，然后从被观察者身(NetTask)上拉取数据。既然提到观察者模式，我们和jdk自带的观察者模式对比，相信很多人都在android代码中使用jdk封装好的类把，那么我就随便举个例子。

public abstract class BaseTask extends Observable{

public static final int NET_ERROR = -1;
public static final int PARSER_ERROR = -2;
public static final int PARAMS_ERROR = -3;
public static final int SUCCESSFUL = 0;
public static final int NOT_VALUE = -4;

protected int resultCode = NET_ERROR;

public int getResultCode(){
return resultCode;
}

public void runTask(final Bundle bundle){

new AsyncTask<Void, Void, Integer>() {

@Override
protected Integer doInBackground(Void... params) {
// TODO Auto-generated method stub
WorkInBackground(bundle);
return resultCode;
}

@Override
protected void onPostExecute(Integer result) {
WorkInUI(result);
BaseTask.this.setChanged();
BaseTask.this.notifyObservers();
}

}.execute();
}

protected abstract void WorkInBackground(Bundle bundle);
protected abstract void WorkInUI(Integer result);

}

我们写个任务的基类来继承Observable，当任务执行完了之后，通知观察者们去取，我们只要在Activity中实现Observer，实现方法update就可以完成数据的更新。

public class SimpleActivity extends Activity implements Observer{

@Override
protected void onCreate(Bundle savedInstanceState) {
...
}

@Override
public void update(Observable observable, Object data) {
//完成数据更新
}

}

这种实现方式虽然也能达到和EventBus相同的作用，但是和我们的EventBus相比，就更能突出我们EventBus最大的优势就是解耦和，将业务和视图分离。只要在BaseAcitity中注册，解绑。通过注解我们就能很随意的获取到任何回调的结果，更为任性的是我们想在哪个线程中运行就再哪个线程中运行，并且接收事件的先后顺序也可以自定义。代码写出来更加的优雅，灵活。也许，你可能会说，性能相比呢？确实，我们的EventBus用用到了大量的缓存和反射，但是3.0后，已经做了很大改变了，我们可以通过apt预编译找到订阅着，避免了运行期间的反射处理解析。更为重要的是，牺牲一点内存达到代码的解耦，以后维护起来更加的方便，我觉得还是有所需要的。最后和Otto框架来比较，你就会知道EventBus框架是多么的优秀了！