EventBus使用
EventBus是一个事件发布/订阅的框架，它能够简化各组件，异步线程和主线程的通信，EventBus代码简洁，它将事件发布和订阅充分解耦。本文是基于eventbus:3.0.0分析。
EventBus引入与使用
在gradle文件下添加依赖：
dependencies {
implementation 'org.greenrobot:eventbus:3.0.0'
}
事件订阅操作
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
EventBus.getDefault().register(this);
}
在这里register可以传递的参数为Object对象。
事件发布操作
EventBus.getDefault().post("str");
//粘性事件
EventBus.getDefault().postSticky("str");
监听EventBus事件的Event回调事件
//默认是POSTING，不用切换线程
@Subscribe(threadMode = ThreadMode.MAIN)
public void test(String str) {
Toast.makeText(mContext, "我是超人", Toast.LENGTH_SHORT).show();
}
//粘性事件的回掉
@Subscribe(threadMode = ThreadMode.MAIN, sticky = true)
public void test(String str) {
Toast.makeText(mContext, "我是超人", Toast.LENGTH_SHORT).show();
}
记得解绑EventBus的操作
@Override
protected void onDestroy() {
super.onDestroy();
EventBus.getDefault().unregister(this);
}
以上就是EventBus的简单使用，可以看到使用方式非常的简单，能够将订阅和发布操作充分的解耦。
上面分别用到了两种方式：粘性事件和普通事件，粘性事件是在事件发布之后，再注册也能够收到回调。
当然EventBus的使用不仅仅只有这些，在这里我们就不一一举例了，有兴趣的可以自行了解，我们主要看看EventBus的原理。
我们现在来看看EventBus的操作原理：
EventBus操作原理
通过上面的介绍，我们大概知道EventBus主要由事件发布，事件订阅构成。
我们先看看EventBus的事件发布/事件订阅的关系图：
eventbus_process.png
EventBus时间发布，订阅流程
eventbus_call.png
Register(订阅者)会调用register方法订阅事件，然后Publisher(事件发布者)通过post接口发布事件，通过EventBus处理之后会执行Subscriber(订阅者)的onEvent事件。
EventBus各类的关系：
eventbus_relation.png
EventBusBuilder：使用构建者模式，用于创建EventBus对象；
SubscriberMethodFinder:订阅者响应函数信息存储和查找类，由 HashMap 缓存，以 ${subscriberClassName} 为 key，SubscriberMethod 对象为元素的 ArrayList 为 value。
SubscriberMethod:订阅者事件响应函数信息，包括响应方法、线程 Mode、事件类型、优先级、黏粘性以及一个用来比较 SubscriberMethod 是否相等的特征值 methodString。
Subscription：订阅信息
HandlerPoster：事件主线程处理，对应ThreadMode.MainThread。继承自 Handler，enqueue 函数将事件放到队列中，并利用 handler 发送 message，handleMessage 函数从队列中取事件，invoke 事件响应函数处理。
BackgroundPoster：事件 Background 处理，对应ThreadMode.BackgroundThread，继承自 Runnable。enqueue 函数将事件放到队列中，并调用线程池执行当前任务，在 run 函数从队列中取事件，invoke 事件响应函数处理。与 AsyncPoster.java 不同的是，BackgroundPoster 中的任务只在同一个线程中依次执行，而不是并发执行。
AsyncPoster：事件异步线程处理，对应ThreadMode.Async，继承自 Runnable。enqueue 函数将事件放到队列中，并调用线程池执行当前任务，在 run 函数从队列中取事件，invoke 事件响应函数处理。
PendingPost：订阅者和事件信息实体类，并含有同一队列中指向下一个对象的指针。通过缓存存储不用的对象，减少下次创建的性能消耗。
EventBus源码分析
创建EventBus的getDefault（）方法：
static volatile EventBus defaultInstance;
/** 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;
}
这里使用的是双重检查的单例模式，并且使用了volatile修饰，前一个对volatile修饰的变量的写操作在后一个volatile的读操作之前，这样会保持变量在内存中的同步。
EventBus构造函数
public EventBus() {
this(DEFAULT_BUILDER);
}
EventBus(EventBusBuilder builder) {
//订阅信息的集合
subscriptionsByEventType = new HashMap<>();
typesBySubscriber = new HashMap<>();
//粘性集合
stickyEvents = new ConcurrentHashMap<>();
//mainThreadPoster 主线程，backgroundPoster后台线程，asyncPoster 异步线程
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;
}
使用EventBusBuilder来初始化一些配置信息，这样可以将配置解耦出去。
EventBus register（注册）
我们先看看订阅流程：
eventbus_register.png
register()方法的实现：
public void register(Object subscriber) {
Class<?> subscriberClass = subscriber.getClass();
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}
传递过来的订阅者，通过findSubscriberMethods方法找到所有的订阅的方法，之后遍历返回的所有的订阅方法进行订阅。
我们来看看找到订阅者的所有订阅方法的findSubscriberMethods（）方法：
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
/*
* private static final Map<Class<?>, List<SubscriberMethod>> METHOD_CACHE
* 从缓存中查找SubscriberMethod集合
*/
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
if (subscriberMethods != null) {
return subscriberMethods;
}
//是否忽略注apt解处理器生成的信息
if (ignoreGeneratedIndex) {
//通过反射获取subscriberMethods
subscriberMethods = findUsingReflection(subscriberClass);
} else {
//通过apt注解器生成的信息获取subscriberMethods,
//如果没有配置MyEventBusIndex，依然通过通过反射获取subscriberMethods
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变量默认为false，并且在findUsingInfo（）方法中也有findUsingReflection（）的调用，我们先进入到findUsingInfo（）：
private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
//创建FindState类
FindState findState = prepareFindState();
//FindState 初始化
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里面的clazz进行重新赋值
findState.moveToSuperclass();
}
//返回订阅方法的集合
return getMethodsAndRelease(findState);
}
首先创建了一个FindState对象并做相应的初始化操作，FindState类中对查找的状态值做了一些封装（查找结果集合subscriberMethods，anyMethodByEventType，subscriberClassByMethodKey，methodKeyBuilder订阅类subscriberClass,事件对象clazz，订阅信息subscriberInfo以及用来标记是否需要进行父类的查看查找的skipSuperClasses），然后通过getSubscriberInfo（findState）方法去找到FindState类里面的subsciberinfo对象，对subscriberinfo进行判空，如果不为空就去通过getSubscriberMethods（）方法返回一个array数组，最后遍历数组，将subscriberMethod给保存到findState的subscriberMehtods集合里面。反之，通过反射来查找订阅方法，同样的也保存在findState的subscriberMehtods集合里，然后对findState里面进行赋值，最后返回订阅方法的集合。
static class FindState {
final List<SubscriberMethod> subscriberMethods = new ArrayList<>();
final Map<Class, Object> anyMethodByEventType = new HashMap<>();
final Map<String, Class> subscriberClassByMethodKey = new HashMap<>();
final StringBuilder methodKeyBuilder = new StringBuilder(128);
Class<?> subscriberClass;
Class<?> clazz;
boolean skipSuperClasses;
SubscriberInfo subscriberInfo;
//...
}
当findState.subscriberInfo == null的时候，会调用反射去拿到订阅方法，现在我们看看findUsingReflectionInSingleClass（）方法：
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标注
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");
}
}
}
findUsingReflectionInSingleClass（）方法中主要使用了反射获取订阅者的所有的方法，然后根据方法的类型，参数和注解来找到订阅方法，找到之后，同前面一样，将订阅方法等信息保存在findState里面。
查到订阅方法集合之后，我们在来看看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);
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();
for (int i = 0; i <= size; i++) {
//按照优先级添加到subscriptions里面
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);
}
}
}
在这里会先去通过(subscriber, subscriberMethod)创建一个subscription订阅信息类，做一下subscriptions集合的初始工作，已经判断是否重复注册，然后按照优先级将新创建的subscription订阅信息添加到subscriptions集合里面，然后添加订阅事件到订阅集合，最后判断订阅者方法是不是粘性事件。至此，以上分析就完成了订阅者的注册操作。
EventBus post（发送）
post发布流程图
EventBus_post.png
先看看post（）方法：
/** Posts the given event to the event bus. */
public void post(Object event) {
//PostingThreadState保存着Event事件队列和线程状态信息
PostingThreadState postingState = currentPostingThreadState.get();
//获取event事件队列
List<Object> eventQueue = postingState.eventQueue;
eventQueue.add(event);
if (!postingState.isPosting) {
//设置isMainThread参数
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;
}
}
}
首先获取PostingThreadState 的状态，从而获取eventQueue，并添加event事件，最后循环发送处理队列中的所有事件。
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
//event事件继承，
if (eventInheritance) {
//查找所有事件并放入Map<Class<?>, List<Class<?>>> eventTypesCache = new HashMap<>()集合中
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) {
Log.d(TAG, "No subscribers registered for event " + eventClass);
}
if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
eventClass != SubscriberExceptionEvent.class) {
post(new NoSubscriberEvent(this, event));
}
}
}
判断是否是eventInheritance为true，true表示向上查找事件的父类。它的默认值为true，可以通过在EventBusBuilder中来进行配置。然后再调用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 {
//对event事件进行处理，订阅
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);
}
}
我们看看这几种方式的具体实现：
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;
}
}
}
BackGroundPoster可以看到它是用线程执行的。
/**
* Posts events in background.
*
* @author Markus
*/
final class BackgroundPoster implements Runnable {
private final PendingPostQueue queue;
private final EventBus eventBus;
private volatile boolean executorRunning;
BackgroundPoster(EventBus eventBus) {
this.eventBus = eventBus;
queue = new PendingPostQueue();
}
public void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
synchronized (this) {
queue.enqueue(pendingPost);
if (!executorRunning) {
executorRunning = true;
eventBus.getExecutorService().execute(this);
}
}
}
@Override
public void run() {
try {
try {
while (true) {
PendingPost pendingPost = queue.poll(1000);
if (pendingPost == null) {
synchronized (this) {
// Check again, this time in synchronized
pendingPost = queue.poll();
if (pendingPost == null) {
executorRunning = false;
return;
}
}
}
eventBus.invokeSubscriber(pendingPost);
}
} catch (InterruptedException e) {
Log.w("Event", Thread.currentThread().getName() + " was interruppted", e);
}
} finally {
executorRunning = false;
}
}
}
BackgroundPoster 继承自 Runnable。enqueue 函数将事件放到队列中，并调用线程池执行当前任务，在 run 函数从队列中取事件，invoke 事件响应函数处理。与 AsyncPoster.java 不同的是，BackgroundPoster 中的任务只在同一个线程中依次执行，而不是并发执行。
AsyncPoster ：
class AsyncPoster implements Runnable {
private final PendingPostQueue queue;
private final EventBus eventBus;
AsyncPoster(EventBus eventBus) {
this.eventBus = eventBus;
queue = new PendingPostQueue();
}
public void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
queue.enqueue(pendingPost);
eventBus.getExecutorService().execute(this);
}
@Override
public void run() {
PendingPost pendingPost = queue.poll();
if(pendingPost == null) {
throw new IllegalStateException("No pending post available");
}
eventBus.invokeSubscriber(pendingPost);
}
}
AsyncPoster 事件异步线程处理，对应ThreadMode.Async，继承自 Runnable。enqueue 函数将事件放到队列中，并调用线程池执行当前任务，在 run 函数从队列中取事件，invoke 事件响应函数处理。
根据主线程，后台线程，异步线程分别对事件进行处理。无论哪种方式最后都会执行eventBus.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);
}
}
到这里大概的分析就已经结束了。
