Android 框架学习2：源码分析 EventBus 3.0 如何实现事件总线

ゝ一世哀愁。 2022-06-14 07:47 265阅读 0赞
  • Go beyond yourself rather than beyond others.

上篇文章 深入理解 EventBus 3.0 之使用篇 我们了解了 EventBus 的特性以及如何使用，这篇文章我们来揭开它神秘的面纱，看看在繁华的背后究竟是怎样的沧桑。

读完本文你将了解：

    • 注解修饰订阅方法
    • 编译时处理注解生成索引

    • 创建 EventBus

      • 最关键的两个属性
      • SubscriberMethod 订阅方法信息
      • Subscription 事件订阅总体
      • SubscriberMethodFinder 订阅方法查找类

    • 注册事件

      • 拿到类中使用 Subscribe 注解修饰的订阅方法
      • 保存订阅方法
      • 注册流程图

    • 取消注册

      • 取消注册普通事件
      • 取消注册粘性事件
      • 取消注册流程图

    • 发送事件

      • 将要发送的事件添加到当前线程的发送队列
      • 遍历处理事件队列中的事件发送前的准备工作
      • 找到订阅该事件的所有订阅者
      • 发送事件给订阅者
      • 一个反射调用方法和三种发送者
      • HandlerPoster 主线程的事件发送者
      • BackgroundPoster 单一子线程发送者
      • AsyncPoster 异步发送者
      • 发送流程图
    • 总结
    • 带有注释的源码地址
    • Thanks

注解修饰订阅方法

EventBus 使用的 Subscribe 注解如下：

  1. @Documented
  2. @Retention(RetentionPolicy.RUNTIME)
  3. @Target({ElementType.METHOD})
  4. public @interface Subscribe {
  6.     ThreadMode threadMode() default ThreadMode.POSTING;
  7.     boolean sticky() default false;
  8.     int priority() default 0;
  9. }

可以看到，由于后面需要在运行时反射获取方法信息，这个注解的 Retention 为运行时，同时只用于修饰方法。

默认 ThreadMode 为 POSTING，即与发送消息者在统一线程。

编译时处理注解生成索引

在注册 EventBus 之前，我们需要创建事件实体类，以及创建订阅方法，比如这样：

  1. @Subscribe(threadMode = ThreadMode.POSTING, priority = 1)
  2. public void readMessage(MessageEvent event) {
  3.     mTvEventInfo.setText("\n" + event.getMessage() + ", " + event.getTime());
  4. }

订阅方法使用 @Subscribe 修饰，在编译时，EventBus 的注解处理器 EventBusAnnotationProcessor 会处理这个方法，然后生成一个实现 SubscriberInfoIndex 接口的类。

SubscriberInfoIndex 代码如下：

  1. public interface SubscriberInfoIndex {
  2.     SubscriberInfo getSubscriberInfo(Class<?> subscriberClass);
  3. }
  4. public interface SubscriberInfo {
  5.     Class<?> getSubscriberClass();    //订阅事件的 Class 对象
  7.     SubscriberMethod[] getSubscriberMethods();    //这个类中的订阅方法数组
  9.     SubscriberInfo getSuperSubscriberInfo();    //父类的订阅信息
  11.     boolean shouldCheckSuperclass();
  12. }

这个接口定义了生成索引需要实现的关键方法，即通过一个 Class 对象获取这个类中的订阅方法数组，父类的订阅信息等等。

注解处理器的代码我们挑重点看一下：

  1. private void createInfoIndexFile(String index) {
  2.     BufferedWriter writer = null;
  3.     try {
  4.         JavaFileObject sourceFile = processingEnv.getFiler().createSourceFile(index);
  5.         int period = index.lastIndexOf('.');
  6.         String myPackage = period > 0 ? index.substring(0, period) : null;
  7.         String clazz = index.substring(period + 1);
  8.         writer = new BufferedWriter(sourceFile.openWriter());
  9.         if (myPackage != null) {
  10.             writer.write("package " + myPackage + ";\n\n");
  11.         }
  12.         writer.write("import org.greenrobot.eventbus.meta.SimpleSubscriberInfo;\n");
  13.         writer.write("import org.greenrobot.eventbus.meta.SubscriberMethodInfo;\n");
  14.         writer.write("import org.greenrobot.eventbus.meta.SubscriberInfo;\n");
  15.         writer.write("import org.greenrobot.eventbus.meta.SubscriberInfoIndex;\n\n");
  16.         writer.write("import org.greenrobot.eventbus.ThreadMode;\n\n");
  17.         writer.write("import java.util.HashMap;\n");
  18.         writer.write("import java.util.Map;\n\n");
  19.         writer.write("/** This class is generated by EventBus, do not edit. */\n");
  20.         writer.write("public class " + clazz + " implements SubscriberInfoIndex {\n");
  21.         writer.write("    private static final Map<Class<?>, SubscriberInfo> SUBSCRIBER_INDEX;\n\n");
  22.         writer.write("    static {\n");
  23.         writer.write("        SUBSCRIBER_INDEX = new HashMap<Class<?>, SubscriberInfo>();\n\n");
  24.         writeIndexLines(writer, myPackage);
  25.         writer.write("    }\n\n");
  26.         writer.write("    private static void putIndex(SubscriberInfo info) {\n");
  27.         writer.write("        SUBSCRIBER_INDEX.put(info.getSubscriberClass(), info);\n");
  28.         writer.write("    }\n\n");
  29.         writer.write("    @Override\n");
  30.         writer.write("    public SubscriberInfo getSubscriberInfo(Class<?> subscriberClass) {\n");
  31.         writer.write("        SubscriberInfo info = SUBSCRIBER_INDEX.get(subscriberClass);\n");
  32.         writer.write("        if (info != null) {\n");
  33.         writer.write("            return info;\n");
  34.         writer.write("        } else {\n");
  35.         writer.write("            return null;\n");
  36.         writer.write("        }\n");
  37.         writer.write("    }\n");
  38.         writer.write("}\n");
  39.         //...
  40. }

可以看到是根据一个 index 创建一个索引类，index 就是这个类的包名和类名。

跟着上篇文章 EventBus 3.0 的使用 配置注解处理器，编译后，就会在 build 文件夹中生成你在 gradle 中配置的索引类。

这里写图片描述

看看这个索引类长什么样吧：

  1. public class MyEventBusIndex implements SubscriberInfoIndex {
  2.         //实现了前面提到的接口
  3.     private static final Map<Class<?>, SubscriberInfo> SUBSCRIBER_INDEX;    //保存类和订阅信息的缓存表
  5.     static {
  6.         SUBSCRIBER_INDEX = new HashMap<Class<?>, SubscriberInfo>();
  8.         //记录每个类中的订阅方法信息
  9.         putIndex(new SimpleSubscriberInfo(net.sxkeji.shixinandroiddemo2.activity.eventbus.EventBusRegisterActivity.class,
  10.                 true, new SubscriberMethodInfo[] {
  11.             new SubscriberMethodInfo("readMessageFirst",
  12.                     net.sxkeji.shixinandroiddemo2.activity.eventbus.MessageEvent.class, ThreadMode.POSTING, 5, false),
  13.             new SubscriberMethodInfo("readMessage", net.sxkeji.shixinandroiddemo2.activity.eventbus.MessageEvent.class,
  14.                     ThreadMode.POSTING, 1, false),
  15.         }));
  17.         putIndex(new SimpleSubscriberInfo(net.sxkeji.shixinandroiddemo2.activity.eventbus.EventBusStickyActivity.class,
  18.                 true, new SubscriberMethodInfo[] {
  19.             new SubscriberMethodInfo("readStickyMsg",
  20.                     net.sxkeji.shixinandroiddemo2.activity.eventbus.MessageEvent.class, ThreadMode.MAIN, 0, true),
  21.         }));
  23.     }
  25.     private static void putIndex(SubscriberInfo info) {
  26.         SUBSCRIBER_INDEX.put(info.getSubscriberClass(), info);
  27.     }
  29.     @Override
  30.     public SubscriberInfo getSubscriberInfo(Class<?> subscriberClass) {
  31.         SubscriberInfo info = SUBSCRIBER_INDEX.get(subscriberClass);
  32.         if (info != null) {
  33.             return info;
  34.         } else {
  35.             return null;
  36.         }
  37.     }
  38. }

可以看到，编译时生成的索引类将我们项目中使用 @Subscribe 修饰的订阅方法信息和所在 Class 都保存在 SUBSCRIBER_INDEX 这个映射表中，这样运行时就可以调用 getSubscriberInfo 方法根据 Class 对象获得订阅信息。

正是这个索引类的存在，在一定程度上减轻了之前版本使用反射获取订阅方法信息的性能问题，大大提高了 EventBus 的运行效率。

创建 EventBus

使用 EventBus 第一步基本都是 EventBus.getDefault() 获取默认配置的 EventBus，先来看看它的源码。

  1. /** Convenience singleton for apps using a process-wide EventBus instance. */
  2. public static EventBus getDefault() {
  3.     if (defaultInstance == null) {
  4.         synchronized (EventBus.class) {
  5.             if (defaultInstance == null) {
  6.                 defaultInstance = new EventBus();
  7.             }
  8.         }
  9.     }
  10.     return defaultInstance;
  11. }

单例模式调用 EventBus 的构造函数:

  1. private static final EventBusBuilder DEFAULT_BUILDER = new EventBusBuilder();
  2. public EventBus() {
  3.     this(DEFAULT_BUILDER);
  4. }
  6. EventBus(EventBusBuilder builder) {
  7.     subscriptionsByEventType = new HashMap<>();
  8.     typesBySubscriber = new HashMap<>();
  9.     stickyEvents = new ConcurrentHashMap<>();
  10.     mainThreadPoster = new HandlerPoster(this, Looper.getMainLooper(), 10);
  11.     backgroundPoster = new BackgroundPoster(this);
  12.     asyncPoster = new AsyncPoster(this);
  13.     indexCount = builder.subscriberInfoIndexes != null ? builder.subscriberInfoIndexes.size() : 0;
  14.     subscriberMethodFinder = new SubscriberMethodFinder(builder.subscriberInfoIndexes,
  15.             builder.strictMethodVerification, builder.ignoreGeneratedIndex);
  16.     logSubscriberExceptions = builder.logSubscriberExceptions;
  17.     logNoSubscriberMessages = builder.logNoSubscriberMessages;
  18.     sendSubscriberExceptionEvent = builder.sendSubscriberExceptionEvent;
  19.     sendNoSubscriberEvent = builder.sendNoSubscriberEvent;
  20.     throwSubscriberException = builder.throwSubscriberException;
  21.     eventInheritance = builder.eventInheritance;
  22.     executorService = builder.executorService;
  23. }

getDefault() 使用默认的 EventBusBuilder 构建了一个 EventBus，从上面的代码我们可以看到，EventBus 类中包含的关键属性如下：

  • eventTypesCache ：保存事件类型的缓存，HashMap
  • subscriptionsByEventType：事件类型与订阅者列表的映射，HashMap
  • typesBySubscriber：订阅的类与订阅的事件关联列表，HashMap
  • stickyEvents：粘性事件，ConcurrentHashMap
  • currentPostingThreadState：当前发送线程的状态，ThreadLocal
  • mainThreadPoster：主线程的消息发送者，Handler
  • backgroundPoster：子线程的消息发送者，Runnable
  • asyncPoster：异步消息发送者，Runnable
  • subscriberMethodFinder：订阅方法查找
  • executorService：线程池

这里介绍几个关键的类。

最关键的两个属性

EventBus 中最关键的两个属性是：

  1. //  事件与对应的订阅者关联列表，
  2. private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;
  3. //  订阅者与订阅的事件关联列表
  4. private final Map<Object, List<Class<?>>> typesBySubscriber;

它俩分别保存了事件与对应的订阅者关联列表、订阅者与订阅事件的关联列表。

在后面的 注册、解除注册中，最后都是在操作它们。注册时添加、解除注册时移除。

至于为什么要用两种映射关系，我们在后面看完整个流程总结。

SubscriberMethod 订阅方法信息

SubscriberMethod 这个类保存了我们写的订阅方法的信息，包括以下信息：

  1. final Method method;    //订阅方法本身
  2. final ThreadMode threadMode;    //执行在哪个线程
  3. final Class<?> eventType;    //事件类型？
  4. final int priority;    //优先级
  5. final boolean sticky;    //是否为粘性事件
  6. String methodString;    //方法名称，用于内部比较两个订阅方法是否相等

Subscription 事件订阅总体

  1. final class Subscription {
  3.     final Object subscriber;
  4.     final SubscriberMethod subscriberMethod;
  5.     /**
  6.      * Becomes false as soon as {
  7.     @link EventBus#unregister(Object)} is called, which is checked by queued event delivery
  8.      * {
  9.     @link EventBus#invokeSubscriber(PendingPost)} to prevent race conditions.
  10.      */
  11.     volatile boolean active;
  13.     Subscription(Object subscriber, SubscriberMethod subscriberMethod) {
  14.         this.subscriber = subscriber;
  15.         this.subscriberMethod = subscriberMethod;
  16.         active = true;
  17.     }
  18.     //...
  19. }

可以看到 Subscription 中保存了类与其中一个订阅方法，在后面发送事件时，是以它为基本事件发送单位。

SubscriberMethodFinder 订阅方法查找类

这个类用来查找和缓存该类中订阅方法信息的类，关键代码如下：

  1. private static final Map<Class<?>, List<SubscriberMethod>> METHOD_CACHE = new ConcurrentHashMap<>();
  2. //根据 Class 对象获取订阅方法列表
  3. List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
  5.     List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
  6.     if (subscriberMethods != null) {
  7.         return subscriberMethods;
  8.     }
  10.     if (ignoreGeneratedIndex) {   //使用 EventBus.getDefault() 时默认是 false
  11.         subscriberMethods = findUsingReflection(subscriberClass); //反射获取 class 中的订阅的方法
  12.     } else {
  13.         subscriberMethods = findUsingInfo(subscriberClass); //从 index 中查询订阅方法
  14.     }
  15.     if (subscriberMethods.isEmpty()) {
  16.         throw new EventBusException("Subscriber " + subscriberClass
  17.                 + " and its super classes have no public methods with the @Subscribe annotation");
  18.     } else {
  19.         METHOD_CACHE.put(subscriberClass, subscriberMethods);
  20.         return subscriberMethods;
  21.     }
  22. }

findSubscriberMethods 方法的的作用就是：根据 Class 对象获取订阅方法列表,流程如下：

  • 先从 METHOD_CACHE 这个 map 中看有没有，有就直接返回
  • 没有的话就根据 ignoreGeneratedIndex 是否为 true 决定使用反射还是使用注解生成的索引来获取订阅方法
  • 找到后将这次的结果放到 METHOD_CACHE 中，便于下次快速查找

由于 EventBus.getDefault() 使用的默认 Builder 中 ignoreGeneratedIndex 是 false 的，所以我们直接看后一种获取订阅方法 findUsingInfo():

  1. /**
  2.  * 先从注解处理器生成的 index 文件中查找，找不到就使用反射获取
  3.  * @param subscriberClass
  4.  * @return
  5.  */
  6. private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
  7.     FindState findState = prepareFindState();
  8.     findState.initForSubscriber(subscriberClass);
  9.     while (findState.clazz != null) {
  10.         findState.subscriberInfo = getSubscriberInfo(findState);    //去 index 文件中获取
  11.         if (findState.subscriberInfo != null) {
  12.             SubscriberMethod[] array = findState.subscriberInfo.getSubscriberMethods();
  13.             //将订阅方法转存到 findState 中
  14.             for (SubscriberMethod subscriberMethod : array) {
  15.                 if (findState.checkAdd(subscriberMethod.method, subscriberMethod.eventType)) {
  16.                     findState.subscriberMethods.add(subscriberMethod);
  17.                 }
  18.             }
  19.         } else {    //index 文件不存在，还是得去反射获取
  20.             findUsingReflectionInSingleClass(findState);
  21.         }
  22.         findState.moveToSuperclass();
  23.     }
  24.     return getMethodsAndRelease(findState);
  25. }
  26. /**
  27.  * 从 index 中获取
  28.  * @param findState
  29.  * @return
  30.  */
  31. private SubscriberInfo getSubscriberInfo(FindState findState) {
  32.     //已经有订阅信息，并且父订阅信息的 Class 就是当前 Class
  33.     if (findState.subscriberInfo != null && findState.subscriberInfo.getSuperSubscriberInfo() != null) {
  34.         SubscriberInfo superclassInfo = findState.subscriberInfo.getSuperSubscriberInfo();  
  35.         if (findState.clazz == superclassInfo.getSubscriberClass()) {
  36.             return superclassInfo;
  37.         }
  38.     }
  39.     //如果有 index 文件，就去 index 文件中查找，没有就返回 null
  40.     if (subscriberInfoIndexes != null) {
  41.         for (SubscriberInfoIndex index : subscriberInfoIndexes) {
  42.             SubscriberInfo info = index.getSubscriberInfo(findState.clazz);
  43.             if (info != null) {
  44.                 return info;
  45.             }
  46.         }
  47.     }
  48.     return null;
  49. }

可以看到 findUsingInfo () 方法中使用复用的 FindState 对象来查询订阅信息，先从注解处理器生成的 index 文件中查找，找不到就使用反射获取。

注册事件

拿到 EventBus 实例后，就可以注册订阅方法，普通事件和粘性事件的注册都是调用：EventBus.getDefault().register(this);。

看下这个方法代码：

  1. public void register(Object subscriber) {
  2.     Class<?> subscriberClass = subscriber.getClass();
  3.     //通过 index 文件或者反射来获得该类中的订阅方法
  4.     List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
  5.     synchronized (this) {
  6.         for (SubscriberMethod subscriberMethod : subscriberMethods) {
  7.             ////遍历这个类中的所有订阅方法，挨个添加到订阅列表
  8.             subscribe(subscriber, subscriberMethod);
  9.         }
  10.     }
  11. }

可以看到注册主要分为两步：

  1. 通过 index 文件或者反射来获得该类中的订阅方法
  2. 遍历这个类中的所有订阅方法，挨个添加到订阅列表

拿到类中使用 Subscribe 注解修饰的订阅方法

register() 方法中调用了我们前面介绍的 subscriberMethodFinder.findSubscriberMethods() 方法得到注册类中使用 Subscribe 注解修饰的方法。

由于 EventBus.getDefault() 默认其实使用的还是反射，这里还是看下它的反射获取订阅方法：

  1. /**
  2.  * 使用反射获取
  3.  * @param findState
  4.  */
  5. private void findUsingReflectionInSingleClass(FindState findState) {
  6.     Method[] methods;
  7.     try {
  8.         // This is faster than getMethods, especially when subscribers are fat classes like Activities
  9.         methods = findState.clazz.getDeclaredMethods();
  10.     } catch (Throwable th) {
  11.         methods = findState.clazz.getMethods();
  12.         findState.skipSuperClasses = true;
  13.     }
  14.     for (Method method : methods) {     //遍历这个类里的所有方法
  15.         int modifiers = method.getModifiers();
  16.         //检查是否是 public 非 static 非抽象的方法
  17.         if ((modifiers & Modifier.PUBLIC) != 0 && (modifiers & MODIFIERS_IGNORE) == 0) {
  18.             Class<?>[] parameterTypes = method.getParameterTypes();
  19.             if (parameterTypes.length == 1) {   //参数必须只能有一个
  20.                 Subscribe subscribeAnnotation = method.getAnnotation(Subscribe.class);
  21.                 if (subscribeAnnotation != null) {
  22.                     Class<?> eventType = parameterTypes[0];
  23.                     if (findState.checkAdd(method, eventType)) {
  24.                         ThreadMode threadMode = subscribeAnnotation.threadMode();
  25.                         findState.subscriberMethods.add(new SubscriberMethod(method, eventType, threadMode,
  26.                                 subscribeAnnotation.priority(), subscribeAnnotation.sticky()));
  27.                     }
  28.                 }
  29.             } else if (strictMethodVerification && method.isAnnotationPresent(Subscribe.class)) {
  30.                 String methodName = method.getDeclaringClass().getName() + "." + method.getName();
  31.                 throw new EventBusException("@Subscribe method " + methodName +
  32.                         "must have exactly 1 parameter but has " + parameterTypes.length);
  33.             }
  34.         } else if (strictMethodVerification && method.isAnnotationPresent(Subscribe.class)) {
  35.             String methodName = method.getDeclaringClass().getName() + "." + method.getName();
  36.             throw new EventBusException(methodName +
  37.                     " is a illegal @Subscribe method: must be public, non-static, and non-abstract");
  38.         }
  39.     }
  40. }

先使用反射获取这个类中的所有方法，不清楚 getDeclaredMethods() 和 getMethods() 区别的同学可以看这篇文章：反射获取成员方法

然后遍历类中的所有方法，将符合条件的订阅方法保存到方法参数 findState 中。

如果遇到使用 Subscribe 注解修饰但是不满足这个条件的方法：

  • public 、非 static 、非抽象的
  • 参数只有一个，多了少了都不行

就会抛出异常。

保存订阅方法

下一步就是：遍历这个类中的所有订阅方法，挨个添加到订阅列表，调用 subscribe() 方法：

  1. private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
  2.     Class<?> eventType = subscriberMethod.eventType;
  3.     Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
  4.     //获取这个事件对应的订阅者列表
  5.     CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
  6.     if (subscriptions == null) {    //如果之前没有这个事件的订阅者，添加进去
  7.         subscriptions = new CopyOnWriteArrayList<>();
  8.         subscriptionsByEventType.put(eventType, subscriptions);
  9.     } else {    //如果之前这个方法已经订阅过这个事件，再次订阅就报错
  10.         if (subscriptions.contains(newSubscription)) {
  11.             throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
  12.                     + eventType);
  13.         }
  14.     }
  16.     int size = subscriptions.size();
  17.     for (int i = 0; i <= size; i++) {
  18.         //遍历这个事件的订阅方法，调整它在订阅列表中的顺序，高优先级的放
  19.         if (i == size || subscriberMethod.priority > subscriptions.get(i).subscriberMethod.priority) {
  20.             subscriptions.add(i, newSubscription);
  21.             break;
  22.         }
  23.     }
  25.     //保存这个类与订阅事件的映射关系
  26.     List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
  27.     if (subscribedEvents == null) {
  28.         subscribedEvents = new ArrayList<>();
  29.         typesBySubscriber.put(subscriber, subscribedEvents);
  30.     }
  31.     subscribedEvents.add(eventType);
  33.     //如果是粘性事件，立即发送出去
  34.     if (subscriberMethod.sticky) {
  35.         if (eventInheritance) {
  36.             Set<Map.Entry<Class<?>, Object>> entries = stickyEvents.entrySet();
  37.             for (Map.Entry<Class<?>, Object> entry : entries) {
  38.                 Class<?> candidateEventType = entry.getKey();
  39.                 if (eventType.isAssignableFrom(candidateEventType)) {
  40.                     Object stickyEvent = entry.getValue();
  41.                     checkPostStickyEventToSubscription(newSubscription, stickyEvent);
  42.                 }
  43.             }
  44.         } else {
  45.             Object stickyEvent = stickyEvents.get(eventType);
  46.             checkPostStickyEventToSubscription(newSubscription, stickyEvent);
  47.         }
  48.     }
  49. }

方法中具体做了什么可以看注解，这里简单概括一下，subscribe() 方法主要做了这些事：

  • 判断这个方法之前是否订阅过这个事件，重复订阅会抛出异常
  • 之前没订阅过的话就把方法的参数，也就是我们订阅的事件 与当前类以及该事件的订阅方法的映射关系保存到 EventBus.subscriptionsByEventType 中
  • 然后按照订阅方法的优先级，调整它在 subscriptionsByEventType 中的顺序，高优先级的放后面
  • 然后保存这个类与订阅事件的映射关系，即前面介绍过的，EventBus 的另外一个关键属性，typesBySubscriber
  • 最后如果是粘性事件的话，立即发送出去，发送的方法我们后续分析

注册流程图

总结一下，注册的流程图如下：

这里写图片描述

结合前面的介绍，现在看这个图是否清晰了不少呢。

取消注册

注册事件订阅后，记得在不需要的时候解除注册，尤其是在 Activity Fragment 这种主线程运行的组件，在它们的 onDestory() 时记得取消注册。

在之前的版本如果不取消注册，下次再注册就会导致注册两回，一旦有发送消息，就会回调两回。我就遇到过这种问题。

从前面注册的源码我们可以看到，现在 3.0 重复注册 EventBus 会抛异常，因此还是要注意。

普通事件和粘性事件解除注册调用的方法不同，我们分开介绍。

取消注册普通事件

取消注册普通事件调用的是 EventBus.getDefault().unregister(this) 方法，看一下源码：

  1. /**
  2.  * 解除注册很简单，就从两个属性中移除当前类
  3.  * @param subscriber
  4.  */
  5. public synchronized void unregister(Object subscriber) {
  6.     List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
  7.     if (subscribedTypes != null) {
  8.         //找到这个类中所有的订阅方法，挨个取消注册
  9.         for (Class<?> eventType : subscribedTypes) {
  10.             unsubscribeByEventType(subscriber, eventType);
  11.         }
  12.         //移除这个订阅记录
  13.         typesBySubscriber.remove(subscriber);
  14.     } else {
  15.         Log.w(TAG, "Subscriber to unregister was not registered before: " + subscriber.getClass());
  16.     }
  17. }
  18. private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
  19.     //获取订阅这个事件的订阅者列表
  20.     List<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
  21.     if (subscriptions != null) {
  22.         int size = subscriptions.size();    //只调用一次 subscriptions.size()，比在循环里每次都调用 subscriptions.size() 好
  23.         for (int i = 0; i < size; i++) {
  24.             Subscription subscription = subscriptions.get(i);
  25.             if (subscription.subscriber == subscriber) {    //将当前订阅者从订阅这个事件的订阅者列表中删除
  26.                 subscription.active = false;
  27.                 subscriptions.remove(i);
  28.                 i--;
  29.                 size--;
  30.             }
  31.         }
  32.     }
  33. }

可以看到，取消注册普通事件很简单，就是从前面强调的，EventBus 的两个属性中移除当前类、类中注册方法、注册事件的关联数据。

取消注册粘性事件

取消注册粘性事件调用的是 EventBus.getDefault().removeStickyEvent() 和 removeAllStickyEvents() 方法，看一下源码：

  1. private final Map<Class<?>, Object> stickyEvents;
  2. //传入的是 Class 对象
  3. public <T> T removeStickyEvent(Class<T> eventType) {
  4.     synchronized (stickyEvents) {
  5.         return eventType.cast(stickyEvents.remove(eventType));
  6.     }
  7. }
  8. //传入的是事件对象
  9. public boolean removeStickyEvent(Object event) {
  10.     synchronized (stickyEvents) {
  11.         Class<?> eventType = event.getClass();
  12.         Object existingEvent = stickyEvents.get(eventType);
  13.         if (event.equals(existingEvent)) {
  14.             stickyEvents.remove(eventType);
  15.             return true;
  16.         } else {
  17.             return false;
  18.         }
  19. }
  20. //移除所有粘性事件
  21. public void removeAllStickyEvents() {
  22.     synchronized (stickyEvents) {
  23.         stickyEvents.clear();
  24.     }
  25. }

可以看到，removeStickyEvent()方法有两个重载，参数分别为 Class 对象和事件对象，但最终都是从 EventBus.stickyEvents 属性中移除当前事件。

取消注册流程图

这里写图片描述

可以看到，其中主要就是对两个关键属性的操作。

原文地址：张拭心的博客

发送事件

终于来到发送事件这一步了，胜利就在眼前！

发送普通事件调用的是 EventBus.getDefault().post(...) 方法；
发送粘性事件调用的是 EventBus.getDefault().postSticky(...) 方法。

  1. public void postSticky(Object event) {
  2.     synchronized (stickyEvents) {
  3.         stickyEvents.put(event.getClass(), event);
  4.     }
  5.     // Should be posted after it is putted, in case the subscriber wants to remove immediately
  6.     post(event);
  7. }

可以看到，发送粘性事件和普通事件的区别就是它先把事件放到了 stickyEvents 属性中，然后再调用的 post()：

  1. /** Posts the given event to the event bus. */
  2. public void post(Object event) {
  3.     //获取当前线程的发送状态
  4.     PostingThreadState postingState = currentPostingThreadState.get();
  6.     List<Object> eventQueue = postingState.eventQueue;
  7.     //将当前事件添加到队列中
  8.     eventQueue.add(event);
  10.     if (!postingState.isPosting) {
  11.         postingState.isMainThread = Looper.getMainLooper() == Looper.myLooper();
  12.         postingState.isPosting = true;
  13.         if (postingState.canceled) {
  14.             throw new EventBusException("Internal error. Abort state was not reset");
  15.         }
  16.         try {
  17.             while (!eventQueue.isEmpty()) {
  18.                 //将队列中的事件都发送出去
  19.                 postSingleEvent(eventQueue.remove(0), postingState);
  20.             }
  21.         } finally { //发送完重置状态
  22.             postingState.isPosting = false;
  23.             postingState.isMainThread = false;
  24.         }
  25.     }
  26. }

可以看到，发送主要分为几步：

  1. 获取当前线程的发送状态
  2. 将当前事件添加到队列中
  3. 将队列中的事件挨个发送出去

将要发送的事件添加到当前线程的发送队列

EventBus.currentPostingThreadState 属性保存了每个线程的相关状态，是一个 ThreadLocal:

  1. private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {
  2.     @Override
  3.     protected PostingThreadState initialValue() {
  4.         return new PostingThreadState();
  5.     }
  6. };
  7. /**
  8.  * 线程的发送状态
  9.  */
  10. final static class PostingThreadState {
  11.     final List<Object> eventQueue = new ArrayList<Object>();    //当前线程的事件队列
  12.     boolean isPosting;
  13.     boolean isMainThread;
  14.     Subscription subscription;
  15.     Object event;
  16.     boolean canceled;
  17. }

它保存了每个线程中的事件队列 eventQueue，是否正在遍历事件的 isPosting，是否为主线程 isMainThread 等信息。

关于 ThreadLocal 你可以查看 这篇文章中的介绍。

将要发送的事件添加到事件队列中后，只要当前线程不是正在遍历发送，就开始发送！

遍历处理事件队列中的事件，发送前的准备工作

  1. try {
  2.             while (!eventQueue.isEmpty()) {
  3.                 //将队列中的事件都发送出去
  4.                 postSingleEvent(eventQueue.remove(0), postingState);
  5.             }
  6.         } finally { //发送完重置状态
  7.             postingState.isPosting = false;
  8.             postingState.isMainThread = false;
  9.         }

可以看到，每个事件都要通过 postSingleEvent() 方法的处理才能发出去。

  1. private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
  2.     Class<?> eventClass = event.getClass();
  3.     boolean subscriptionFound = false;
  4.     if (eventInheritance) {
  5.         List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);    //找到这个事件的所有父类和接口
  6.         int countTypes = eventTypes.size();
  7.         for (int h = 0; h < countTypes; h++) {
  8.             Class<?> clazz = eventTypes.get(h);
  9.             subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
  10.         }
  11.     } else {
  12.         subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
  13.     }
  14.     if (!subscriptionFound) {   //如果没有订阅者，调用失败，根据配置，抛出一个 NoSubscriberEvent 事件
  15.         if (logNoSubscriberMessages) {
  16.             Log.d(TAG, "No subscribers registered for event " + eventClass);
  17.         }
  18.         if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
  19.                 eventClass != SubscriberExceptionEvent.class) {
  20.             post(new NoSubscriberEvent(this, event));
  21.         }
  22.     }
  23. }

由于 EventBus 中有事件继承的概念，所以在 postSingleEvent() 方法中，先找到了这个事件的所有父类和接口，然后把这些亲戚都发出去，调用的是 postSingleEventForEventType() 方法。

找到订阅该事件的所有订阅者

  1. /**
  2.  * 发送该事件给所有订阅者
  3.  * @param event
  4.  * @param postingState
  5.  * @param eventClass 事件的 Class 对象
  6.  * @return 是否调用成功
  7.  */
  8. private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
  9.     CopyOnWriteArrayList<Subscription> subscriptions;
  10.     synchronized (this) {   //拿到事件的订阅者列表                ，数据有可能在操作的同时添加，考虑使用 CopyOnWriteArrayList 与 synchronized
  11.         subscriptions = subscriptionsByEventType.get(eventClass);
  12.     }
  13.     if (subscriptions != null && !subscriptions.isEmpty()) {
  14.         for (Subscription subscription : subscriptions) {
  15.             postingState.event = event;
  16.             postingState.subscription = subscription;
  17.             boolean aborted = false;
  18.             try {
  19.                 postToSubscription(subscription, event, postingState.isMainThread);
  20.                 aborted = postingState.canceled;
  21.             } finally {
  22.                 postingState.event = null;
  23.                 postingState.subscription = null;
  24.                 postingState.canceled = false;
  25.             }
  26.             if (aborted) {
  27.                 break;
  28.             }
  29.         }
  30.         return true;
  31.     }
  32.     return false;
  33. }

可以看到，subscriptionsByEventType 的确是 EventBus 最重要的两个属性之一，在发送消息时，需要通过它拿到订阅该事件的所有订阅者。

然后调用 postToSubscription() 方法挨个给订阅者发送事件。

发送事件给订阅者

订阅方法中会指定被调用时所在的线程 ThreadMode，因此我们在发送事件给订阅方法时，需要在指定的线程里调用它：

  1. /**
  2.  * 根据 threadMode 发送事件给该订阅者
  3.  * @param subscription
  4.  * @param event
  5.  * @param isMainThread
  6.  */
  7. private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
  8.     switch (subscription.subscriberMethod.threadMode) {
  9.         case POSTING:       //直接在当前线程调用
  10.             invokeSubscriber(subscription, event);
  11.             break;
  12.         case MAIN:      //订阅方法要在主线程执行
  13.             if (isMainThread) {     //当前就在主线程，直接调用
  14.                 invokeSubscriber(subscription, event);
  15.             } else {        //入队，handler 发送消息到主线程，在主线程中反射调用
  16.                 mainThreadPoster.enqueue(subscription, event);
  17.             }
  18.             break;
  19.         case BACKGROUND:
  20.             if (isMainThread) {     //放到子线程线程池中执行
  21.                 backgroundPoster.enqueue(subscription, event);
  22.             } else {
  23.                 invokeSubscriber(subscription, event);
  24.             }
  25.             break;
  26.         case ASYNC:     //直接在子线程中执行
  27.             asyncPoster.enqueue(subscription, event);
  28.             break;
  29.         default:
  30.             throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
  31.     }
  32. }

可以看到，对应 EventBus 的四种 ThreadMode，有一个反射调用方法和三种发送者。

一个反射调用方法和三种发送者

反射调用方法为：invokeSubscriber()，先看它的代码：

  1. /**
  2.  * 反射调用订阅方法
  3.  * @param subscription
  4.  * @param event
  5.  */
  6. void invokeSubscriber(Subscription subscription, Object event) {
  7.     try {
  8.         subscription.subscriberMethod.method.invoke(subscription.subscriber, event);
  9.     } catch (InvocationTargetException e) {
  10.         handleSubscriberException(subscription, event, e.getCause());
  11.     } catch (IllegalAccessException e) {
  12.         throw new IllegalStateException("Unexpected exception", e);
  13.     }
  14. }

它的代码很简单，就是直接使用反射调用订阅者的方法，参数是调用对象和参数。

如果订阅方法要求被调用的线程和发送者在同一线程，就会直接调用 invokeSubscriber() 方法。

否则就交给三种发送者：

  1. HandlerPoster mainThreadPoster：主线程的事件发送者
  2. BackgroundPoster backgroundPoster：单一子线程发送者
  3. AsyncPoster asyncPoster：异步发送者

HandlerPoster 主线程的事件发送者

主线程调用，根据我们的经验，应该是使用 Handler 把事件以消息的形势发到主线程吧，看看mainThreadPoster 的源码是不是：

  1. final class HandlerPoster extends Handler {
  4.     private final PendingPostQueue queue;
  5.     private final int maxMillisInsideHandleMessage;
  6.     private final EventBus eventBus;
  7.     private boolean handlerActive;
  9.     //...
  10. }

HandlerPoster 中持有一个 PendingPostQueue 对象，它是 PendingPost 双向链表，具体内容

  1. /**
  2.  * 待发送消息的双向链表，生产者-消费者模型, 出队 wait - 入队 nofityAll
  3.  */
  4. final class PendingPostQueue {
  5.     private PendingPost head;
  6.     private PendingPost tail;
  8.     synchronized void enqueue(PendingPost pendingPost) {
  9.         if (pendingPost == null) {
  10.             throw new NullPointerException("null cannot be enqueued");
  11.         }
  12.         if (tail != null) {
  13.             tail.next = pendingPost;
  14.             tail = pendingPost;
  15.         } else if (head == null) {
  16.             head = tail = pendingPost;
  17.         } else {
  18.             throw new IllegalStateException("Head present, but no tail");
  19.         }
  20.         notifyAll();
  21.     }
  23.     synchronized PendingPost poll() {
  24.         PendingPost pendingPost = head;
  25.         if (head != null) {
  26.             head = head.next;
  27.             if (head == null) {
  28.                 tail = null;
  29.             }
  30.         }
  31.         return pendingPost;
  32.     }
  34.     synchronized PendingPost poll(int maxMillisToWait) throws InterruptedException {
  35.         if (head == null) {
  36.             wait(maxMillisToWait);  //最大等待超时
  37.         }
  38.         return poll();
  39.     }
  41. }

而 PendingPost 就是对要发送事件和订阅者的又一次封装，和之前查找订阅者使用的 FindState 一样，都是为了避免在多个方法传递参数时传递参数太多，比如包一起用一个对象得了。此外还做了对象复用池，对象使用后不销毁，加入复用池中，就不用创建太多对象。

PendingPost 代码如下：

  1. /**
  2.  * 等待发送的事件队列，就是一个对象复用池
  3.  */
  4. final class PendingPost {
  5.     private final static List<PendingPost> pendingPostPool = new ArrayList<PendingPost>();  //对象复用池
  7.     Object event;
  8.     Subscription subscription;
  9.     PendingPost next;
  10. }

我们回去看 HandlerPoster 的关键代码：

  1. {
  2.     //...
  3.     void enqueue(Subscription subscription, Object event) {
  4.         PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);   //为什么要包裹成 PendingPost 呢？
  5.         synchronized (this) {
  6.             queue.enqueue(pendingPost);     //事件入队
  7.             if (!handlerActive) {
  8.                 handlerActive = true;
  9.                 if (!sendMessage(obtainMessage())) {    //入一次队，发送一个空消息
  10.                     throw new EventBusException("Could not send handler message");
  11.                 }
  12.             }
  13.         }
  14.     }

可以看到， HandlerPoster 在接收到新的要发送事件后，将要发送事件包成 PendingPost 入队，然后给 Looper 发一个空消息，触发调用 handleMessage() 。

  1. @Override
  2.     public void handleMessage(Message msg) {        //这里运行在主线程
  3.         boolean rescheduled = false;
  4.         try {
  5.             long started = SystemClock.uptimeMillis();
  6.             while (true) {
  7.                 PendingPost pendingPost = queue.poll();    //取消息
  8.                 if (pendingPost == null) {    //如果当前队列中没有消息，就阻塞等待
  9.                     synchronized (this) {
  10.                         // Check again, this time in synchronized
  11.                         pendingPost = queue.poll();    
  12.                         if (pendingPost == null) {
  13.                             handlerActive = false;
  14.                             return;
  15.                         }
  16.                     }
  17.                 }
  18.                 eventBus.invokeSubscriber(pendingPost);    //应该也是反射
  19.                 long timeInMethod = SystemClock.uptimeMillis() - started;
  20.                 if (timeInMethod >= maxMillisInsideHandleMessage) {
  21.                     if (!sendMessage(obtainMessage())) {
  22.                         throw new EventBusException("Could not send handler message");
  23.                     }
  24.                     rescheduled = true;
  25.                     return;
  26.                 }
  27.             }
  28.         } finally {
  29.             handlerActive = rescheduled;
  30.         }
  31.     }

由于EventBus 中的 HandlerPoster 对象初始化中使用的是主线程的 Looper：

  1. mainThreadPoster = new HandlerPoster(this, Looper.getMainLooper(), 10);     //主线程发送者

因此这里的 handleMessage() 就已经运行在主线程了。

在这个方法中我们不停地从队列中取出 PendingPost ，然后调用eventBus.invokeSubscriber(pendingPost) 发出去：

  1. void invokeSubscriber(PendingPost pendingPost) {
  2.     Object event = pendingPost.event;
  3.     Subscription subscription = pendingPost.subscription;
  4.     PendingPost.releasePendingPost(pendingPost);
  5.     if (subscription.active) {
  6.         invokeSubscriber(subscription, event);
  7.     }
  8. }

可以看到，取出其中的事件和订阅者后，最终还是调用我们的反射调用方法。

BackgroundPoster 单一子线程发送者

BackgroundPoster 代码如下：

  1. /**
  2.  * Posts events in background.
  3.  *  使用线程池循环执行任务
  4.  * @author Markus
  5.  */
  6. final class BackgroundPoster implements Runnable {
  8.     private final PendingPostQueue queue;
  9.     private final EventBus eventBus;
  11.     private volatile boolean executorRunning;
  13.     BackgroundPoster(EventBus eventBus) {
  14.         this.eventBus = eventBus;
  15.         queue = new PendingPostQueue();
  16.     }
  18.     public void enqueue(Subscription subscription, Object event) {
  19.         PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
  20.         synchronized (this) {    //一个线程中可能会同时入队多个事件，需要同步
  21.             queue.enqueue(pendingPost); //入队，执行任务
  22.             if (!executorRunning) {
  23.                 executorRunning = true;
  24.                 eventBus.getExecutorService().execute(this);    //使用线程池执行
  25.             }
  26.         }
  27.     }
  28.     //...
  30. }

可以看到，BackgroundPoster 是一个 Runnable ，在入队事件后，调用eventBus.getExecutorService().execute(this) 分配一个线程执行自己，触发调用 run() 方法:

  1. {
  2.     //...
  3.     @Override
  4.     public void run() {
  5.         try {
  6.             try {
  7.                 while (true) {
  8.                     PendingPost pendingPost = queue.poll(1000);     //有可能阻塞
  9.                     if (pendingPost == null) {
  10.                         synchronized (this) {
  11.                             // Check again, this time in synchronized
  12.                             pendingPost = queue.poll();
  13.                             if (pendingPost == null) {
  14.                                 executorRunning = false;
  15.                                 return;
  16.                             }
  17.                         }
  18.                     }
  19.                     eventBus.invokeSubscriber(pendingPost);
  20.                 }
  21.             } catch (InterruptedException e) {
  22.                 Log.w("Event", Thread.currentThread().getName() + " was interruppted", e);
  23.             }
  24.         } finally {
  25.             executorRunning = false;
  26.         }
  27.     }

BackgroundPoster.run() 类似 HandlerPoster.handleMessage() ，不停地从队列中取出 PendingPost ，然后调用eventBus.invokeSubscriber(pendingPost) ，最终还是通过反射调用订阅方法。

由于 BackgroundPoster 是在一个线程中不停地发射调用订阅方法，如果某个订阅方法太耗时，会影响其他订阅方法的调用。

AsyncPoster 异步发送者

AsyncPoster 代码很简单：

  1. class AsyncPoster implements Runnable {
  4.     private final PendingPostQueue queue;
  5.     private final EventBus eventBus;
  7.     AsyncPoster(EventBus eventBus) {
  8.         this.eventBus = eventBus;
  9.         queue = new PendingPostQueue();
  10.     }
  12.     public void enqueue(Subscription subscription, Object event) {
  13.         PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
  14.         queue.enqueue(pendingPost);
  15.         eventBus.getExecutorService().execute(this);
  16.     }
  17.     //...
  18. }

可以看到，AsyncPoster 和 BackgroundPoster 有一些相似，都是 Runnable，都通过 EventBus 的线程池中分配一个线程执行。

不同的是 BackgroundPoster在 run() 方法中串行调用队列中的所有订阅方法，为了保证在调用过程中另外入队情况的同步性，入队方法和 run() 方法都需要使用 synchronized 同步块做同步处理。

而 AsyncPoster.run() 方法就简单多了，每次只执行一个订阅方法的调用。有新的订阅方法就新建一个 AsyncPoster.run()。

因此官方建议要控制 ThreadMode 为 ASYNC的订阅方法数量，避免创建大量线程。

  1. @Override
  2.     public void run() {
  3.         PendingPost pendingPost = queue.poll();
  4.         if(pendingPost == null) {
  5.             throw new IllegalStateException("No pending post available");
  6.         }
  7.         eventBus.invokeSubscriber(pendingPost);
  8.     }

发送流程图

发送的流程看似复杂，其实经过前面的介绍，结合下面这幅图，也很清晰了：

这里写图片描述

总结

本篇文章根据源码详细地了解了 EventBus 3.0 的源码，经过前面对源码的分析，现在看这个类图理解了吧：

这里写图片描述

一句话来总结 EventBus 3.0 的实现原理：

  • 注册时有两种方式获取类中的订阅方法 
  1.  *  从编译时注解处理器生成的索引文件中获取
  2.  *  反射遍历类中的方法，检查注解是否合格
  • 发送时，根据不同的 ThreadMode 调用不同的 Poster 在不同的线程反射调用订阅方法

就是酱紫！！

读源码最好是打着断点走一遍，然后边走边写注释，最后整理成文章。

在这个过程中发现了很多小细节是我之前所忽略的，比如一些并发容器的使用，对象复用池，还有一些状态的判断，具体学习心得还有很多，我们下一篇文章总结。

带有注释的源码地址

Thanks

本来打算自己画流程图和类图的，但发现前辈画的已经很完美的，而且我也有点困，就借用一下吧 -。-。

本文中的三张流程图引用自 老司机教你 “飙” EventBus 3 ，表示感谢！

最后类图引用自 EventBus 源码解析，表示感谢！

https://segmentfault.com/a/1190000005089229
http://www.jianshu.com/p/f057c460c77e
http://liuwangshu.cn/application/eventbus/2-eventbus-sourcecode.html
http://blog.csdn.net/z609933542/article/details/50953166

发表评论

表情：
评论列表 （有 0 条评论，265人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Android事件总线EventBus 3.X.X

    为了简化并且更加高质量地在Activity、Fragment、Thread和Service等之间的通信，同时解决组件之间 高耦合的同时仍能继续高效地通信，事件总线设计出现了。提

    悠悠悠悠/ 0 赞/ 80 阅读