前言
EventBus使用和源码都比较简单，适合新手上手源码阅读。代码基于3.2.0版本，implementation 'org.greenrobot:eventbus:3.2.0'。
主要功能
简单使用流程：在需要订阅消息的地方（activity或者fragment），定义处理消息的方法，并用@Subscribe修饰，同时要在create注册，在destroy解注册，然后在合适的地方，调用post发送该类型的消息，就能在处理消息的地方收到发送的消息。
其中有下面几个主要功能：
1.获取EventBus对象
2.注册
3.解注册
4.发送消息
5.接收消息
val eventBus = EventBus.getDefault()
eventBus.register(this)
eventBus.unregister(this)
eventBus.post("post")
eventBus.postSticky("postSticky")
@Subscribe(threadMode = ThreadMode.MAIN)
fun back(str: String) {
}
获取EventBus对象
一般使用getDefault()方法，这是一个使用双重校验锁方式实现的单例模式，返回一个EventBus对象，源码如下：
static volatile EventBus defaultInstance;
// Convenience singleton for apps using a process-wide EventBus instance.
public static EventBus getDefault() {
EventBus instance = defaultInstance;
if (instance == null) {
synchronized (EventBus.class) {
instance = EventBus.defaultInstance;
if (instance == null) {
instance = EventBus.defaultInstance = new EventBus();
}
}
}
return instance;
}
可以看到EventBus中有两个构造方法
public EventBus() {
this(DEFAULT_BUILDER);
}
EventBus(EventBusBuilder builder) {
logger = builder.getLogger();
subscriptionsByEventType = new HashMap<>();
......一些初始化操作
eventInheritance = builder.eventInheritance;
executorService = builder.executorService;
}
EventBus的对象创建使用了建造者模式（Builder模式），除了使用默认的builder构造EventBus，还可以使用自定义的builder，EventBus提供了一个静态方法生成Builder对象，而在Builer中又提供了方法生成EventBus对象，代码如下：
//静态方法生产Builder
public static EventBusBuilder builder() {
return new EventBusBuilder();
}
//builder对象生成EventBus对象
public EventBus build() {
return new EventBus(this);
}
综上，获取EventBus对象有如下两种方式：
//1进程单例方式获取，使用默认Builder
val eventBus1 = EventBus.getDefault()
//2用户在默认Builder的基础上，可以按照需求自定义Builder参数
val eventBus2 = EventBus.builder()
.ignoreGeneratedIndex(true)
......自定义Builder参数
.logger(null)
.build()
注册
总注册流程
register(this)方法源码如下：
public void register(Object subscriber) {
//1.获取传入参数的class对象
//2.通过subscriberMethodFinder找到传入对象类中所有使用@Subscribe修饰的方法
//3.把找到的方法存起来
Class<?> subscriberClass = subscriber.getClass();//1
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);//2
//3
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}
找订阅方法
先看订阅方法包装类SubscriberMethod，有以下几个属性
final Method method;//具体的方法
final ThreadMode threadMode;//指定线程模式
final Class<?> eventType;//事件类型，方法的参数的class对象
final int priority;//优先级
final boolean sticky;//是否是粘性消息
/** Used for efficient comparison */
String methodString;
在用注解@Subscribe修饰方法时，有三个参数可以选，和上面的属性对应，源码如下：
public @interface Subscribe {
ThreadMode threadMode() default ThreadMode.POSTING;
boolean sticky() default false;
int priority() default 0;
}
threadMode属性，表示指定的线程模式，默认使用ThreadMode.POSTING，ThreadMode是一个枚举类，有以下五种类型：ThreadMode.POSTING、ThreadMode.MAIN、ThreadMode.MAIN_ORDERED、ThreadMode.BACKGROUND、ThreadMode.ASYNC。每个枚举的具体含义后面再说。
然后是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()) {
//该类下面没有注解@Subscribe修饰的订阅方法，不可注册，抛出异常
throw new EventBusException("Subscriber " + subscriberClass
+ " and its super classes have no public methods with the @Subscribe annotation");
} else {
//把找到的订阅方法缓存到METHOD_CACHE并返回所有订阅方法，METHOD_CACHE是一个ConcurrentHashMap对象，支持并发
METHOD_CACHE.put(subscriberClass, subscriberMethods);
return subscriberMethods;
}
}
看一看找订阅方法的具体逻辑，如果ignoreGeneratedIndex为true，就调用findUsingReflection(subscriberClass)，否则调用findUsingInfo(subscriberClass)。参数含义是 是否忽略生成索引，索引是EventBus在3.0新加的功能，目的是解决直接使用反射查找订阅方法带来的性能问题，我们先看不使用索引的逻辑，即ignoreGeneratedIndex为true，关于索引的内容放后面再说。
findUsingReflection源码如下：
private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
//1.获取FindState实例，
//使用FindState[] FIND_STATE_POOL数组存放findState实例，数组长度固定为4
//如果数组里面没有实例，新建一个findState
//如果数组里面有实例，获取对象使用，避免新建对象带来的消耗，并把池里面的该位置置空，
//和下面的getMethodsAndRelease方法相反，getMethodsAndRelease会把使用的findState清空数据，然后放入池中，以便下次使用
SubscriberMethodFinder.FindState findState = prepareFindState();
//2.根据订阅类subscriberClass初始化findState
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
//3.通过反射方式，找到订阅者下面的所有订阅方法
findUsingReflectionInSingleClass(findState);
//4.再去他的父类里面找订阅方法
findState.moveToSuperclass();
}
//5.通过findState实例，获取所有找到的订阅方法，并把findState清空数据后放入池中
return getMethodsAndRelease(findState);
}
其中关键代码是3处的findUsingReflectionInSingleClass，通过反射找订阅方法，源码如下：
private void findUsingReflectionInSingleClass(SubscriberMethodFinder.FindState findState) {
//1.反射获取所有方法
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
try {
methods = findState.clazz.getMethods();
} catch (LinkageError error) { // super class of NoClassDefFoundError to be a bit more broad...
String msg = "Could not inspect methods of " + findState.clazz.getName();
if (ignoreGeneratedIndex) {
msg += ". Please consider using EventBus annotation processor to avoid reflection.";
} else {
msg += ". Please make this class visible to EventBus annotation processor to avoid reflection.";
}
throw new EventBusException(msg, error);
}
findState.skipSuperClasses = true;
}
//2.遍历所有方法，找到满足条件的订阅方法
for (Method method : methods) {
int modifiers = method.getModifiers();
//方法是public，且非static，非abstract
if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
Class<?>[] parameterTypes = method.getParameterTypes();
//只能有一个参数
if (parameterTypes.length == 1) {
Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
//方法被注解Subscribe修饰
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");
}
}
}
代码看起来比较多，其实逻辑很清晰，一看就明白是在做什么。
把找到的方法存起来
遍历找到的订阅方法，调用subscribe方法保存，subscribe源码如下：
// Must be called in synchronized block
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
Class<?> eventType = subscriberMethod.eventType;
//使用订阅者类和订阅方法，生成订阅类Subscription实例
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
//根据事件类型，获取所有的订阅subscriptions，如果为空，新建并放入到subscriptionsByEventType这个map里面，是事件和CopyOnWriteArrayList<Subscription>的映射
CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions == null) {
subscriptions = new CopyOnWriteArrayList<>();
subscriptionsByEventType.put(eventType, subscriptions);
} else {
//如果subscriptions不为空，且newSubscription实例已经包含在subscriptions中，说明同一个订阅者的同一个事件重复注册了，抛出异常
if (subscriptions.contains(newSubscription)) {
throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
+ eventType);
}
}
//按照订阅方法的优先级，把订阅newSubscription添加到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;
}
}
//把订阅事件添加到对应订阅者的订阅事件list
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);
}
}
}
至此，注册流程完成。
解注册
unregister(this)方法源码如下：
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，提示该订阅者没有注册
logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
}
}
private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
//根据订阅事件，获取所有的订阅
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--;
}
}
}
}
逻辑比较清晰简单，只要看懂了注册流程，解注册自然而然就明白是怎么回事了。
发送消息
发送消息分为普通消息和粘性消息，先看看普通消息的发送
post()源码如下：
public void post(Object event) {
//获取当前线程下的postingState，currentPostingThreadState是一个ThreadLocal<PostingThreadState>实例，保证拿到的postingState是通过线程隔离的
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;
}
}
}
再看看postSingleEvent的逻辑，都会走到postSingleEventForEventType方法，然后走到postToSubscription方法，下面是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，不进行线程切换，接收消息和发送消息在同一个线程；
MAIN，在主线程接收消息，如果发送线程是主线程，直接调用invokeSubscriber，如果是子线程，把消息放到mainThreadPoster队列；
MAIN_ORDERED，在主线程接收消息，和MAIN的不同之处在于，优先把消息放到mainThreadPoster队列，只有当mainThreadPoster为空是才会直接调用invokeSubscriber；
BACKGROUND，在子线程接收消息，如果发送线程是主线程，把消息放到子线程队列backgroundPoster，如果是子现场，直接调用invokeSubscriber；
ASYNC，在子线程接收消息，独立于发送线程和主线程，直接把消息放到asyncPoster队列。
上述流程中，最关键的两个点是invokeSubscriber和asyncPoster队列（他和mainThreadPoster以及backgroundPoster原理一样，看一个就行），下面看看这两部分的代码。
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);
}
}
逻辑简单清晰，直接通过反射调用订阅方法。
asyncPoster队列原理是，把消息放在队列里面，然后通过线程池拿到子线程后，在子线程获取消息队列里的消息，然后再执行上面的invokeSubscriber。每来一条消息就开一个线程处理，消息之间是并发的。
backgroundPoster原理和asyncPoster类似，区别是增加了同步逻辑，用来保证消息顺序执行，上一条消息执行完成再执行下一条消息。
mainThreadPoster和backgroundPoster类似，都有加同步，然后通过handler发送消息到主线程，再在主线程调用invokeSubscriber。
说完了post，在来看一下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);
}
逻辑比较简单清晰，先把事件存到map里面，然后再调用post，因为是用的map存，所以同一个事件只会有最新的消息。
以上，就是发送消息以及接受消息的所有流程。
索引
使用
首先在app的build.gradle中加入配置:
android {
defaultConfig {
javaCompileOptions {
annotationProcessorOptions {
// 根据项目实际情况，指定辅助索引类的名称和包名
arguments = [ eventBusIndex : 'com.af.etest.MyEventBusIndex' ]
}
}
}
}
dependencies {
compile 'org.greenrobot:eventbus:3.2.0'
// 引入注解处理器
annotationProcessor 'org.greenrobot:eventbus-annotation-processor:3.2.0'
}
然后新建两个activity，增加订阅方法，如下：
public class MainActivity extends AppCompatActivity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
EventBus eventBus = EventBus.builder().addIndex(new MyEventBusIndex()).build();
eventBus.register(this);
}
@Subscribe(threadMode = ThreadMode.ASYNC,priority = 100,sticky = true)
public void callback1(String e) {
}
@Subscribe
public void callback2(Activity activity) {
}
}
public class TestActivity extends AppCompatActivity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_test);
EventBus.getDefault().register(this);
}
@Subscribe
public void testCallback(TestActivity testActivity){
}
}
编译之后，就会自动生成如下的java文件：
//This class is generated by EventBus, do not edit.
public class MyEventBusIndex implements SubscriberInfoIndex {
private static final Map<Class<?>, SubscriberInfo> SUBSCRIBER_INDEX;
static {
SUBSCRIBER_INDEX = new HashMap<Class<?>, SubscriberInfo>();
putIndex(new SimpleSubscriberInfo(TestActivity.class, true, new SubscriberMethodInfo[] {
new SubscriberMethodInfo("testCallback", TestActivity.class),
}));
putIndex(new SimpleSubscriberInfo(MainActivity.class, true, new SubscriberMethodInfo[] {
new SubscriberMethodInfo("callback1", String.class, ThreadMode.ASYNC, 100, true),
new SubscriberMethodInfo("callback2", android.app.Activity.class),
}));
}
private static void putIndex(SubscriberInfo info) {
SUBSCRIBER_INDEX.put(info.getSubscriberClass(), info);
}
@Override
public SubscriberInfo getSubscriberInfo(Class<?> subscriberClass) {
SubscriberInfo info = SUBSCRIBER_INDEX.get(subscriberClass);
if (info != null) {
return info;
} else {
return null;
}
}
}
基本原理就是在编译时，先获取到所有的订阅方法信息，然后存到指定的SubscriberInfoIndex实现类，以便注册时使用。实现这一功能的代码在这儿
源码
当没有忽略索引的时候，注册流程中，找订阅方法会走到findUsingInfo方法，和上文提过的findUsingReflection方法相比，增加了注释1处的代码，源码如下：
private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
//1
while (findState.clazz != null) {
//通过builder设置的SubscriberInfoIndex实现类，获取到SubscriberInfo，这个实现类就是使用索引那一步，自动生成的类MyEventBusIndex
findState.subscriberInfo = getSubscriberInfo(findState);
if (findState.subscriberInfo != null) {
//通过获取到的SubscriberInfo信息，生成订阅方法
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);
}
和不使用索引相比，索引是在编译的时候就获得了所有的订阅方法信息，而不用在运行时通过反射去获取，大大提升了注册的性能。
其他
另外，EventBus还有几个不常用的api，cancelEventDelivery()取消事件，removeAllStickyEvents()移除所有粘性消息，源码也比较简单清晰，如下：
public void cancelEventDelivery(Object event) {
PostingThreadState postingState = currentPostingThreadState.get();
if (!postingState.isPosting) {
throw new EventBusException(
"This method may only be called from inside event handling methods on the posting thread");
} else if (event == null) {
throw new EventBusException("Event may not be null");
} else if (postingState.event != event) {
throw new EventBusException("Only the currently handled event may be aborted");
} else if (postingState.subscription.subscriberMethod.threadMode != ThreadMode.POSTING) {
throw new EventBusException(" event handlers may only abort the incoming event");
}
postingState.canceled = true;
}
public void removeAllStickyEvents() {
synchronized (stickyEvents) {
stickyEvents.clear();
}
}
最后
对这次的源码阅读做一个小小的总结。
读完源码后，对EventBus的使用基本上全部清楚，包括一些不常使用的功能。另外也对EventBus的整个实现流程，包括获取对象实例，注册，发送消息等都比较清楚。还对这些流程里面的细节，包括找订阅方法的两种方式，发送消息时的线程切换等比较清楚。
虽然EventBus用起来比较简单，源码不多，但是有很多地方值得学习的，比如：
使用Builder模式新建对象；
对一些需要耗时才能生成的对象进行缓存，例如订阅方法，findState实例；
使用队列处理消息；
线程池的使用和线程切换；
在编译期生成文件，避免反射导致的性能问题；