EventBus 源码分析

0. 前言

EventBus 是一款针对Android优化的发布/订阅事件总线。主要功能是替代 Intent , Handler , BroadCastFragmentActivityService ,线程之间传递消息。优点是开销小,代码更优雅,以及将发送者和接收者解耦。此文将对最新的 EventBus 3.0 的源码进行简要的分析。

1. 用法回顾

EventBus 3.0 的用法较之前的版本有所变化,它使用了最近较为流行的注解形式取代以前的 onEvent 开头作为方法名,但使用步骤大致相同,分为三个步骤:

  • 注册与反注册
    在Activity中的 onCreate() 方法中进行EventBus的注册,在 onDestroy() 方法中进行取消注册。

    1
    2
    3
    4
    // 注册
    EventBus.getDefault().register(this);
    // 取消注册
    EventBus.getDefault().unregister(this);
  • 订阅
    在3.0版本中,使用 @Subscribe 注解指定订阅者方法。

    1
    2
    @Subscribe(threadMode = ThreadMode.MAIN)
    public void function(Params p){ }
  • 发布
    发布者通过 post()postSticky() 方法进行发布消息。

    1
    EventBus.getDefault().post(new Params());

注意:

  1. 注册类经笔者测试,必须要是Activity才可以
  2. 订阅者方法可见性不能为 private

2. 从注册入手

2.1 单例

调用 getDefault() 方法可以获取到一个全局的单例对象, 这里采用的是 饿汉式 单例设计模式:

1
2
3
4
5
6
7
8
9
10
public static EventBus getDefault() {
if (defaultInstance == null) {
synchronized (EventBus.class) {
if (defaultInstance == null) {
defaultInstance = new EventBus();
}
}
}
return defaultInstance;
}

2.2 构造方法

单例模式维持了一个唯一的对象,那这个对象包含那些东西呢?接下来让我们看看构造方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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;
}

首先是三张 HashMap ,用来存储订阅信息;然后,有创建了三个 Poster ,这就是用来反馈到订阅者所用的。

2.3 注册订阅

1
2
3
4
5
6
7
8
9
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(...) 方法获取订阅类中包含订阅注解的所有方法,最后将这些方法通过 subscribe() 来实现订阅。

2.4 获取注解方法

为了提高性能,先经过一层存放方法的缓存区,如果缓存区有数据,这直接从缓冲区中获取,没有的话才通过反射机制获取方法,同时将获取到的方法存放一份到缓存区。

1
2
3
4
5
6
7
8
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
...
subscriberMethods = findUsingReflection(subscriberClass);
...
METHOD_CACHE.put(subscriberClass, subscriberMethods);
return subscriberMethods;
}

如果没有缓存数据,则通过反射查找所有包含 @Subscribe 注解的方法, findUsingReflection() 方法会查找继承关系中所有包含注解的方法:

1
2
3
4
5
6
7
8
9
private List<SubscriberMethod> findUsingReflection(Class<?> subscriberClass) {
FindState findState = prepareFindState();
findState.initForSubscriber(subscriberClass);
while (findState.clazz != null) {
findUsingReflectionInSingleClass(findState);
findState.moveToSuperclass();
}
return getMethodsAndRelease(findState);
}

具体的查找过程代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private void findUsingReflectionInSingleClass(FindState findState) {
...
methods = findState.clazz.getDeclaredMethods();
...
for (Method method : methods) {
...
// 获取 `@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()));
}
}
...
}
...
}

代码比较长,大体思路就是先通过字节码利用反射获取所有方法,然后遍历查找包含注解的方法,在检查合法性后添加到 findState.subscriberMethods 列表,列表定义如下:

1
2
3
4
5
6
static class FindState {
final List<SubscriberMethod> subscriberMethods = new ArrayList<>();
final Map<Class, Object> anyMethodByEventType = new HashMap<>();
final Map<String, Class> subscriberClassByMethodKey = new HashMap<>();
...
}

至此,订阅者的方法就找到了,回到 register() 方法,了解下一步的操作。

1
2
3
4
5
6
public void register(Object subscriber) {
...
for (SubscriberMethod subscriberMethod : subscriberMethods) {
subscribe(subscriber, subscriberMethod);
}
}

哈哈,很明显,接下来就是把包含注解的所有方法进行订阅。

3. 订阅

3.1 存放订阅信息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
...
Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
...
subscriptions.add(i, newSubscription);
...
subscribedEvents.add(eventType);
if (subscriberMethod.sticky) {
...
// 获取事件
Object stickyEvent = stickyEvents.get(eventType);
checkPostStickyEventToSubscription(newSubscription, stickyEvent);
}
}

订阅方法中分为两种,一种是立即推送,它被添加至 subscriptions 列表中保存,还有一种 sticky 模式,笔者理解的就是延迟模式,它就调用 checkPostStickyEventToSubscription() 方法。我们先来看后者:

1
2
3
4
5
private void checkPostStickyEventToSubscription(Subscription newSubscription, Object stickyEvent) {
if (stickyEvent != null) {
postToSubscription(newSubscription, stickyEvent, Looper.getMainLooper() == Looper.myLooper());
}
}

这里调用了 postToSubscription() 方法,带有三个参数,分别是包含订阅者方法的订阅信息包装类、延迟事件、是否为主线程。笔者这里想说下第三个参数, Looper.getMainLooper() 方法返回的是主线程的Looper对象,Looper.myLooper() 返回的是当前线程的Looper对象,所以判断这两个Looper对象是否相等就可以判断当前线程是否为主线程了,是不是很巧妙呢!好了,继续前进去看看 postToSubscription() 方法。

3.2 提交订阅

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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);
}
}

提交订阅也是可以大致分为两种,直接调用或者放入构造方法中提到的 Poster

3.2.1 直接调用

1
2
3
4
5
6
7
8
9
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);
}
}

直接使用 Method#invoke() 方法调用即可,不多说。

3.2.2 Poster方式

Poster 有三个,原理都基本一致的,这里就以 mainThreadPoster 来说就好,其它两个就靠读者举一反三了:)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 private final PendingPostQueue queue;

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() 获取到一个 PendingPost 对象,其包含:

1
2
3
4
private PendingPost(Object event, Subscription subscription) {
this.event = event;
this.subscription = subscription;
}

然后将其放在存放 PendingPostPendingPostQueue 队列中,并且将标志位 handlerActive 激活。

1
2
3
4
final class HandlerPoster extends Handler {
void enqueue(Subscription subscription, Object event) { ... }
public void handleMessage(Message msg) { ... }
}

最后利用Handler机制,调用 sendMessage(obtainMessage()) 发送消息,传递到 handleMessage() 中对消息进行处理。

3.3 取出消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@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) {
// 双重检查
pendingPost = queue.poll();
// 如果没有 pendingPost ,则消息处理完毕,退出
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.invokeSubscriber(pendingPost) 处理消息,另外,如果处理超时,还会重新执行 sendMessage(obtainMessage()) 以重试。

3.4 调用订阅者

接下来看 eventBus.invokeSubscriber(pendingPost) 方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
void invokeSubscriber(PendingPost pendingPost) {
Object event = pendingPost.event;
Subscription subscription = pendingPost.subscription;
PendingPost.releasePendingPost(pendingPost);
if (subscription.active) {
invokeSubscriber(subscription, event);
}
}
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);
}
}

思路就是对 pendingPost 解包,取出 subscription ,然后调用其中的方法。但是现在队列中并没有消息,所以上述还不会执行,我们得通过发布者发布消息后才能取出消息执行。

4. 发布者

4.1 发布事件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public void post(Object event) {
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;
}
}
}

该方法中,先是获取到事件队列,然后将参数放入队列,刷新 postingState 后执行事件。

4.2 提交事件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
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) {
Log.d(TAG, "No subscribers registered for event " + eventClass);
}
if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
eventClass != SubscriberExceptionEvent.class) {
post(new NoSubscriberEvent(this, event));
}
}
}

大体就是获取到事件类型,然后调用 postSingleEventForEventType() 方法:

1
2
3
4
5
6
7
8
9
private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
...
for (Subscription subscription : subscriptions) {
postingState.event = event;
postingState.subscription = subscription;
boolean aborted = false;
postToSubscription(subscription, event, postingState.isMainThread);
...
}

同样,先要刷新 postingState 状态,然后调用 postToSubscription() 方法,这个方法是不是有点熟悉呢?

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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);
}
}

对,这又回到了前文中的提交订阅部分了,之后 Poster 取出消息,传递消息,调用订阅者。

5. 总结

EventBus 作为安卓消息传递中的一大神器,单凭笔者这篇博文也是分析不透的,不过希望读者阅读此文后能有一个初步的印象,了解具体的流程,对以后的开发工作有所帮助。