盒子
盒子
文章目录
  1. 概念
  2. EventBus的注册
    1. 索引机制
    2. 反射机制
    3. 订阅
  3. EventBus的发布
  4. EventBus的注销
  5. 总结
  6. 关注

EventBus的工作原理总结

2018.02.27 Rouse android  热度 ℃

概念

EventBus是一个面向于Android与Java的对事件订阅与发布的处理的一种机制。简单的说它就是一种事件处理机制，该项目已经开源。下面是一张图能够很全面的体现它的运行。

这里将EventBus的工作原理分为三步来说明

  1. EventBus的注册与订阅
  2. EventBus的事件发布
  3. EventBus的注销

EventBus的注册

既然是对工作原理进行总结，那么废话不多说，我们直接进入EventBus的代码进行分析

该文章的代码都是基于EventBus3.0.0来进行分析说明

在使用EventBus开始之前都会调用它的注册代码

1
EventBus.getDefault().register(this)

通过getDefault()方法来获取EventBus的实例

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;
}

通过源码发现其实就是一个典型的singleton pattern(单例模式)，保证实例的全局化，减少资源的开销。

进入EventBus的注册代码：

1
2
3
4
5
6
7
8
9
10
public void register(Object subscriber) {
    Class<?> subscriberClass = subscriber.getClass();
    //获取该订阅对象中的所有订阅了的方法
    List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
    synchronized (this) {
        for (SubscriberMethod subscriberMethod : subscriberMethods) {
            subscribe(subscriber, subscriberMethod);
        }
    }
}

在这里有一个subscriberMethodFinder成员字段，通过调用它的findSubscriberMethods方法来获取订阅的方法数组。转到findSubscriberMethods方法中看下它的实现

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

上面关键部分的注释已经写出来了。如果本地缓存中没有，则获取方式根据ignoreGeneratedIndex分为两种。其中ignoreGeneratedIndex默认为false。该字段的值在EventBus实例化的适合默认进行了设置。

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

显而易见是通过builder模式来配置默认的参数值，所以如果需要改变这其中的参数值，可以通过builder来设置

1
2
EventBus eventBus = EventBus.builder().ignoreGeneratedIndex(true).build();
eventBus.register(this);

索引机制

既然默认是false，那么就会调用findUsingInfo方法

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 List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
	//实例化findState
    FindState findState = prepareFindState();
    //初始化findState
    findState.initForSubscriber(subscriberClass);
    while (findState.clazz != null) {
    	//获取订阅者的相关信息并保存在findState中
        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.moveToSuperclass();
    }
    //将获取的订阅方法重新添加到一个新的List集合中，并释放findState中的资源
    return getMethodsAndRelease(findState);
}

这里的关键地方是getSubscriberInfo方法，用来获取订阅者的相关信息。该方法的内部查找是使用index索引。如果该方法还是没有找到，则转向使用反射机制进行获取。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private SubscriberInfo getSubscriberInfo(FindState findState) {
    if (findState.subscriberInfo != null && findState.subscriberInfo.getSuperSubscriberInfo() != null) {
        SubscriberInfo superclassInfo = findState.subscriberInfo.getSuperSubscriberInfo();
        if (findState.clazz == superclassInfo.getSubscriberClass()) {
            return superclassInfo;
        }
    }
    if (subscriberInfoIndexes != null) {
    	//遍历查找订阅者的相关信息
        for (SubscriberInfoIndex index : subscriberInfoIndexes) {
            SubscriberInfo info = index.getSubscriberInfo(findState.clazz);
            if (info != null) {
                return info;
            }
        }
    }
    return null;
}

这里的subscriberInfoIndexes就是之前builder中对应的相同字段属性的值。该集合的值可以通过builder来动态添加

1
2
3
EventBus eventBus = EventBus.builder()
    .addIndex(new MyEventBusAppIndex())
    .addIndex(new MyEventBusLibIndex()).build();

到这里就是全部通过index获取订阅方法的实现。该方法只有在EventBus 3.x以上版本才有，因为没有使用反射机制，所以相对来说该方法的速度自然会比通过反射机制来获取更快。

反射机制

反射机制获取的主要实现是在findUsingReflectionInSingleClass方法中，我们进入瞧瞧它的实现步骤

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
34
35
36
37
38
39
40
41
42
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 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");
        }
    }
}

关键注释已给出，主要是通过反射原理来获取订阅方法的所需信息。同时保存下来以便后续的事件发布。下面是之前反射机制步骤开始的外部调用方法源码

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);
}

主要逻辑跟使用索引方式类似，只是少了对索引方式的查找。

订阅

回到register方法中，订阅者的所以订阅方法已经获取完毕，下一步就开始进行订阅操作，本质是对这些方法进行管理保存。那么直接看subscribe方法

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
  private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
  	//订阅方法中所申明的参数class类型
      Class<?> eventType = subscriberMethod.eventType;
      //对订阅者与订阅者的方法建立联系，进行封装管理
      Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
      //通过EventType来查找关联的subscribetions集合
      CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
      if (subscriptions == null) {
          subscriptions = new CopyOnWriteArrayList<>();
          //保存新的eventType的subscriptions
          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;
          }
      }

//根据订阅者来查找所有的EventType
      List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
      if (subscribedEvents == null) {
          subscribedEvents = new ArrayList<>();
          typesBySubscriber.put(subscriber, subscribedEvents);
      }
      //根据订阅者保存所有的EventType
      subscribedEvents.add(eventType);

//对具有sticky类型的订阅方法进行直接发布处理
      if (subscriberMethod.sticky) {
      	//验证eventType与stickyEvent间的继承关系，默认为true
          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();
                  //判断eventType是否是candidateEventType的父类或者同一种类型
                  if (eventType.isAssignableFrom(candidateEventType)) {
                      Object stickyEvent = entry.getValue();
                      //发布事件
                      checkPostStickyEventToSubscription(newSubscription, stickyEvent);
                  }
              }
          } else {
              Object stickyEvent = stickyEvents.get(eventType);
              //发布事件
              checkPostStickyEventToSubscription(newSubscription, stickyEvent);
          }
      }
  }

订阅的处理主要分为两种情况，其中一种是非sticky类型，根据eventType来保存subscriptions与根据subscribe来保存eventType；另一种是sticky类型，那么最终都是对事件进行直接发布。

EventBus的发布

EventBus的发布是调用post方法，指定发布的Event类型，只要订阅者持有该相关Event类型的方法就都能收到发布的事件。具体使用如下：

1
2
3
EventBus.getDefault().post(new MyEvent());
//或者发布sticky类型的event
EventBus.getDefault().postSticky(new MyEvent());

对于sticky类型的事件发布，本质上还是调用了post的方法，只是增加了对sticky event的同步保存

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

所以我们还是直接看post方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
  public void post(Object event) {
      PostingThreadState postingState = currentPostingThreadState.get();
      List<Object> eventQueue = postingState.eventQueue;
      eventQueue.add(event);

//本地线程是否正在发布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()) {
              	//逐一发布单个event
                  postSingleEvent(eventQueue.remove(0), postingState);
              }
          } finally {
              postingState.isPosting = false;
              postingState.isMainThread = false;
          }
      }
  }

在发布中有一个currentPostingThreadState字段，它是一个ThreadLocal，保证当前线程中只有一个发布队列在进行中。把所以需要发布的event加入到正在进行中的队形中。对该队列中的event进行逐个取出发布。实现方法为postSingleEvent

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
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
    Class<?> eventClass = event.getClass();
    boolean subscriptionFound = false;
    //如果开启了event的继承机制
    if (eventInheritance) {
    	//查找所有的eventType，即包括它的父类与实现的接口
        List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
        int countTypes = eventTypes.size();
        for (int h = 0; h < countTypes; h++) {
            Class<?> clazz = eventTypes.get(h);
            //只要其中一个eventType被找到相应的subscription，就为true
            subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
        }
    } else {
    	//标识该eventType是否有相应的subscription
        subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
    }
    //如果没有找到，则发布一个默认的event
    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));
        }
    }
}

从上面的源码可以看出关键代码是postSingleEventForEventType方法，找到对应的subscriptions，即所以与该eventType相关的订阅者中的订阅方法。

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
private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
    CopyOnWriteArrayList<Subscription> subscriptions;
    synchronized (this) {
    	//通过eventType来找到对应的subscriptions
        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中找到对应eventType的subscriptions，然后转入到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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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);
    }
}

这里根据不同的threadMode进行不同的处理，分别为

  • POSTING：当前线程
  • MAIN：主线程
  • MAIN_ORDERED：主线程排队分发
  • BACKGROUND：后台线程
  • ASYNC：异步线程池

不管是排队分发还是直接分发，最终都会调用invokeSubscriber方法

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

最终的发布实现还是使用了反射的原理，通过反射来定义订阅的方法。到这里EventBus的发布过程就结束了。下面进入最后的EventBus的注销。

EventBus的注销

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public synchronized void unregister(Object subscriber) {
	//通过订阅者查找到所有的eventType
    List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
    if (subscribedTypes != null) {
        for (Class<?> eventType : subscribedTypes) {
        	//注销所有的subscriptions
            unsubscribeByEventType(subscriber, eventType);
        }
        //删除注销的订阅者
        typesBySubscriber.remove(subscriber);
    } else {
        logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
    }
}

复杂的都已经分析完了，相对来说EventBus的注销就简单多了，无非删除对于的资源信息。在unregister中首先通过typesBySubscriber找到订阅者的所以eventType，然后调用unsubscribeByEventType方法来删除所有的subscriptions，最后将订阅者从typesBySubscriber中移除。那么最后再来看下unsubscribeByEventType方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
	//通过eventType找到对应的subscriptions
    List<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
    if (subscriptions != null) {
        int size = subscriptions.size();
        //遍历所有的subscription
        for (int i = 0; i < size; i++) {
            Subscription subscription = subscriptions.get(i);
            if (subscription.subscriber == subscriber) {
            	//设置subscription的状态为非活跃状态，保证正在事件分发时，该subscription不被分发
                subscription.active = false;
                //删除
                subscriptions.remove(i);
                i--;
                size--;
            }
        }
    }
}

逻辑也很简单，更新subscriptionsByEventType中的数据，将注销的订阅者的相关信息移除。到这里EventBus的注销也分析完毕。

总结

EventBus的主要流程是：对需要订阅的类进行register，在register过程中根据ignoreGeneratedIndex字段来分别进行index索引或者反射机制获取订阅者中的所有方法信息，另外如果index索引方法没有获取到对于的方法信息，最后还是会通过反射机制保底获取；然后就是subscribe，将获取到的方法信息信息进行保存；进一步就是post，从保存的方法信息中进行匹配到对于的event方法，通过反射机制的invoke调用匹配的方法；最后对不需要的订阅者进行注销，释放资源。

关注

怪谈时间到了

支持一下
赞赏是一门艺术