Eventbus异步框架源码解析大体流程。
EventBus 工作原理
消息发布和订阅模型
生产者：发布者发布消息
消费者：订阅者注册消息
调度中心：执行消息的分发到位
线程切换
主要就是两个切换，主线程切子线程，子线程切主线程
一个进程内通过Looper.getMainLooper可以顺利切换到主线程
子线程通过是通过主线程执行某个线程，自动切换，也可以借助ThreadLocal缓存对象来切换
参考：Android线程消息机制
EventBus 类图
EventBus：核心类，代表了一个事件总线。实现发布-订阅模式。
SubscriberMethodFinder：寻找订阅者的订阅方法
FindState：寻找订阅方法过程中的缓存Buffer，目的内存复用
SubscriberMethod：订阅者的方法封装类
Subscription：订阅者对象的封装
ThreadMode：线程模型，用来标识不同线程执行消息
PostingThreadState：消息发布状态封装
Poster：不同线程消息发布器
EventBus 订阅-发布时序图
EventBus 订阅-发布流程图
订阅流程源码分析
EventBus.getDefault
获取EventBus单例，
// 提供默认单例 EventBus
EventBus.register
获取订阅者Class
通过反射，获取Class的所有订阅方法集合
遍历订阅方法集合，存储两个Map，Map<订阅事件类型，订阅对象集合>， Map<订阅者，订阅事件集合>
如果订阅的是Stick事件类型，则直接寻找对应事件，然后分发到订阅方法
public void register(Object subscriber) {
SubscriberMethodFinder.findSubscriberMethods
先从内存缓存 Map<订阅对象Class,订阅对象Class的订阅方法> 找
然后通过反射方法获取订阅者Class的所有订阅方法
Map 存储
返回订阅者的订阅方法集合
List
findSubscriberMethods(Class<?> subscriberClass) {
SubscriberMethodFinder.findUsingInfo
通过FIND_STATE_POOL构建FindState，用做找寻订阅方法时的缓存对象
执行反射寻找订阅者的订阅方法，存储到FindState中
循环遍历找订阅者的父类的订阅方法
返回订阅者的订阅方法集合，并释放FindState
private List
findUsingInfo(Class<?> subscriberClass) {
SubscriberMethodFinder.findUsingReflectionInSingleClass
遍历类中所有方法，寻找目标订阅方法，封装成SubscriberMethod存储到findState的subscriberMethods集合中
private void findUsingReflectionInSingleClass(FindState findState) {
EventBus.subscribe
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
发布流程源码分析
EventBus.post
参考：深入理解ThreadLocal
public void post(Object event) {
EventBus.postSingleEvent
根据eventInheritance确认是否需要分发父类Class，最终执行postSingleEventForEventType 进行事件分发
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
EventBus.postSingleEventForEventType
获取订阅对象集合
遍历订阅对象，执行事件分发
private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) { // 获取 订阅对象集合 CopyOnWriteArrayList
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;}
EventBus.postToSubscription
根据不同的 ThreadMode 进行处理
执行线程与目标线程匹配，则直接反射执行订阅方法
不匹配，则先将事件添加到PendingPostQueue，然后通过ExecutorService执行Runnable
private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) { // 根据不同的 ThreadMode 进行处理 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); }}// 执行订阅者的订阅方法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); }}
BackgroundPoster
以BackgroundPoster为例，进行代码分析
构建PendingPost，然后加入PendingPostQueue队列，接着调用线程执行器执行线程
线程执行，循环获取PendingPost，调用EventBus.invokeSubscriber反射执行订阅者的订阅方法
final class BackgroundPoster implements Runnable, Poster { ... // 入队 public void enqueue(Subscription subscription, Object event) { // 将 订阅者和事件封装为一个 PendingPost 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); 轮训取 post，1s没有则停止执行 if (pendingPost == null) { synchronized (this) { // Check again, this time in synchronized pendingPost = queue.poll(); if (pendingPost == null) { executorRunning = false; return; } } } // 最后执行eventBus的invokeSubscriber方法 eventBus.invokeSubscriber(pendingPost); } } ... }}
本文分享自微信公众号 - Android架构师成长之路（gh_07f996f00d9b）。
如有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。
