EventBus原理
EventBus作用
事件总线用于观察者模式，用于通知使用，需要注册，和解注册，订阅要执行的方法，然后post发送消息，就可以在注册的方法中执行
EventBus使用
注册和解注册
//注册
EventBus.getDefault().register(this);
//解注册
EventBus.getDefault().unregister(this);
订阅
@Subscribe(threadMode = ThreadMode.POSTING)
public void onEventPostingAction(Action action) {}
@Subscribe(threadMode = ThreadMode.MAIN)
public void onEventMainAction(Action action){}
@Subscribe(threadMode = ThreadMode.MAIN_ORDERED)
public void onEventMainOrderedAction(Action action){ }
@Subscribe(threadMode = ThreadMode.BACKGROUND)
public void onEventBackgroundAction(Action action) {}
@Subscribe(threadMode = ThreadMode.ASYNC)
public void onEventAsyncAction(Action action) { }
1，ThreadMode.POSTING
事件从哪个线程找那个发的就在哪个执行，默认就是这种执行方式，因为消耗最小，不用切换线程
2， ThreadMode.MAIN
事件执行在主线程
3， ThreadMode.MAIN_ORDERED
事件执行在主线程，事件会排列到队列中依次执行
4，ThreadMode.BACKGROUND
如果发布的线程是主线程那么在后台线程执行，如果不是主线程那么就在发布线程执行，主要用于非耗时后台操作
5，ThreadMode.ASYNC
在单独线程中执行，主要用于后台线程耗时操作，会开启线程池
发送
EventBus.getDefault().post(new Aysn());
粘性事件
粘性事件是什么？
想一种场景就是我们在发送事件的时候并没有注册，那这时候是不是就不能接收到事件了，基于这种场景就产生了粘性事件
在安卓中具体怎么应用呢？
安卓我们都知道页面间传递数据使用Binder，Binder缓存也是有限的，最大
1M，正常的话应该连1M都不到，因为还有一些其他消耗，所以当我在页面间传递数据的时候就可以使用粘性事件了
粘性事件在使用时是需要单独的注册和解注册的，如下
注册在注解处多了一个sticky字段
@Subscribe(sticky = true,threadMode = ThreadMode.ASYNC)
public void onEventAsyncAction(Action action) { }
解注册
Action stickyEvent = EventBus.getDefault().getStickyEvent(Action.class);
// 判断此粘性事件是否存在
if (stickyEvent != null) {
// 若粘性事件存在，将其删除
EventBus.getDefault().removeStickyEvent(stickyEvent);
}
EventBus原理（源码基于EventBus3.1.1）
注册
EventBus.getDefault().register(this);
首先调用的是EventBus的getDefault方法
先讨论一下EventBus这个类，我们接触到的类都是EventBus这就是设计模式中的门面模式，即所有的操作都是通过这个类操作
然后看getDefault()都做了什么
public static EventBus getDefault() {
if(defaultInstance == null) {
Class var0 = EventBus.class;
synchronized(EventBus.class) {
if(defaultInstance == null) {
defaultInstance = new EventBus();
}
}
}
return defaultInstance;
}
可以看到使用的是双检测单例的方式，至于为什么是双检测单例可以百度一下实现单例的方式，以及各种实现的利弊
然后调用的是EventBus中的register(this)，这个this就是我们要注册执行方法所在的类（我们就叫他注册类）
EventBus中的register(Object subscriber)
public void register(Object subscriber) {
Class<?> subscriberClass = subscriber.getClass();
List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
synchronized (this) {
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}
}
调用订阅事件所在类的Class字节类，将字节码传递给SubscriberMethodFinder的findSubscriberMethods()方法
SubscriberMethodFinder中的findSubscriberMethods(Class<?> subscriberClass)
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()) {
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;
}
}
这里它调用了METHOD_CACHE.get(subscriberClass)，那METHOD_CACHE是什么呢，看样子应该是一个缓存，我们看看他的定义
private static final Map<Class<?>, List<SubscriberMethod>> METHOD_CACHE = new ConcurrentHashMap<>();
它是一个同步的ConcurrentHashMap，key是传入的class，value是订阅的需要被调用方法，这里呢主要考虑的是反射调用方法是一件耗费资源的事情如果说在Map都存储好了你要反射的注册事件，那如果有，就直接取缓存的就好了，我们往下看，哎，有一个ignoreGeneratedIndex，这个是什么意思呢？ignoreGeneratedIndex表示是否忽略注解器生成的MyEventBusIndex，默认的话是false，忽略的话直接走反射了，不忽略的话走的是编译时生成的注解文件（如果没有取到注解生成的类也会走反射）
看一下走反射的情况
SubscriberMethodFinder中findUsingReflection(subscriberClass);
private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
findUsingReflectionInSingleClass(findState);
findState.moveToSuperclass();
}
return getMethodsAndRelease(findState);
}
首先调用了 prepareFindState();这个方法主要是从一个注册类的状态池中取出一个注册类的状态对象，然后初始化，这个状态池是默认是4个，池的作用是便于管理，这样把注册类的数据都集中管理在注册类状态池中便于对注册类的管理，如果需要直接在注册类状态池中取出就好了
findState.initForSubscriber(subscriberClass);就是初始化注册类状态
这个while循环是检索所有注册类中注册的方法的，通过findState.moveToSuperclass();向上寻找。这里我们可以看到EventBus如果在父类中注册了方法，那么子类再去注册时父类注册的方法也不会失效的
我们在来看
findUsingReflectionInSingleClass(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对象，通过反射获取所有方法，遍历所有的方法，最外层的判断是判断方法是public，并且不是static，abstract，这里还有两种是bridge or synthetic methods（这两种是java生成class的时候自动编译的）
。
获取方法的参数，参数的成都要为1
获取方法上标记的注解，如果注解不为null，获取方法参数class
if (findState.checkAdd(method, eventType)) 使用用于检测方法中重写的情况，如果重写不继续解析
取出threadMode注册方法执行线程的模式，最后将所有数据封装后创建一个SubscriberMethod，存储到注册状态类的subscriberMethods的成员变量中，这个subscriberMethods是一个ArrayList
方法结束回到调用处findUsingReflection(Class<?> subscriberClass)，return返回getMethodsAndRelease(findState);
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中的subscribeMethod，然后释放findState
到这里反射方式就结束了
另一种情况走的是编译时生成的MyEventBusIndex，这里先不表。
回到SubscriberMethodFinder中的findSubscriberMethods(Class<?> subscriberClass)最后返回的是subscribeMethods
在回到EventBus中的register(Object subscriber)，继续走到for循环调用EventBus中的subscribe(subscriber, subscriberMethod)，第一个参数为要注册的类，第二个为解析出的需要注册的方法
EventBus中的subscribe(subscriber, subscriberMethod)
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
Class<?> eventType = subscriberMethod.eventType;
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++) {
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) {
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);
}
}
}
1，取出参数类型
2，创建Subscription，将注册类和注册方法关联起来
3，subscriptionsByEventType类型是Map<Class<?>, CopyOnWriteArrayList<Subscription>>用于关联方法参数的类型，和注册类和注册方法的关系
4，比较优先级，优先级高的排在前面
5，typesBySubscriber类型 Map<Object, List<Class<?>>> key是注册的类，value是注册方法的参数类型，作用就是关联注册类，和注册方法的参数类型
6，判断是否是黏性事件，stickyEvents类型是Map<Class<?>, Object>，key是注册方法的参数类型，value是黏性事件，最后走的checkPostStickyEventToSubscription(newSubscription, stickyEvent);
private void checkPostStickyEventToSubscription(Subscription newSubscription, Object stickyEvent) {
if (stickyEvent != null) {
postToSubscription(newSubscription, stickyEvent, isMainThread());
}
}
postToSubscription()方法就是用于分发事件使用
发送
最常用的发送有两种
第一种是 public void post(Object event) 正常发送事件
第二种是 public void postSticky(Object event) 发送黏性事件
EventBus的post(Object event)
public void post(Object event) {
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()，currentPostingThreadState定义如下
private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {
@Override
protected PostingThreadState initialValue() {
return new PostingThreadState();
}
};
使用ThreadLocal去存储PostingThreadState对象，PostingThreadState是用来保存线程状态的，还有线程中注册的方法，和注册的类
final static class PostingThreadState {
final List<Object> eventQueue = new ArrayList<>();
boolean isPosting;
boolean isMainThread;
Subscription subscription;
Object event;
boolean canceled;
}
回到public void post(Object event)，取出线程状态postingState，然后把发送事件加入到事件队列中
判断postingState的isPosting是否发送状态，如果没发送，给发送事件的线程状态赋值，是否是主线程，是否正在发送，如果发送事件线程的发送事件队列不为空循环发送事件postSingleEvent(eventQueue.remove(0), postingState);
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
Class<?> eventClass = event.getClass();
boolean subscriptionFound = false;
if (eventInheritance) {
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));
}
}
}
取出发送事件的类型，eventInheritance是否通知父类，关键代码是他会调用
postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass)
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取出subscriptions，然后遍历subscriptions，把事件对象和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 {
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);
}
}
这个方法就可以看到，通过判断线程模式然后执行不同的线程模式，我们依次来看
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);
}
}
直接反射调用，还有三种是分别放入到队列中，分别为mainThreadPoster，asyncPoster，backgroundPoster。除了mainThreadPoster其他两个都是实现了Post和Runnable。他们都是调用了Post接口中的enqueue()方法
mainThreadPoster初始化在build中
mainThreadSupport = builder.getMainThreadSupport();
mainThreadPoster = mainThreadSupport != null ? mainThreadSupport.createPoster(this) : null;
public interface MainThreadSupport {
Poster createPoster(EventBus eventBus);
class AndroidHandlerMainThreadSupport implements MainThreadSupport {
...
@Override
public Poster createPoster(EventBus eventBus) {
return new HandlerPoster(eventBus, looper, 10);
}
}
}
最后返回的是HandlerPoster，在来看HandlerPoster的enqueue()
public 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");
}
}
}
}
PendingPost.obtainPendingPost(subscription, event)把事件转换成了PendingPost类型这里维护了一个对象池，用于减少对象的创建，原理和Message.obtain()原理是一样的，然后sendMessage主线程发送消息，主线程反射调用
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
final class BackgroundPoster implements Runnable, Poster {
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) {
eventBus.getLogger().log(Level.WARNING, Thread.currentThread().getName() + " was interruppted", e);
}
} finally {
executorRunning = false;
}
}
}
enqueue()方法取出PendingPost，然后使用线程池执行，AsyncPoster与BackgroundPoster类似，就不多写了
解注册
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 {
logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
}
}
typesBySubscriber首先按注册类取出，注册方法参数，然后for循环在通过注册类从subscriptionsByEventType取出注册的类和方法对象，一层一层remove
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--;
}
}
}
}
