EventBus原理剖析

作者: pphdsny | 来源:发表于2018-01-22 22:37 被阅读242次

EventBus

EventBus 是一个 Android 事件发布/订阅框架，通过解耦发布者和订阅者简化 Android 事件传递。传统的事件传递方式包括：Handler、BroadCastReceiver、Interface 回调，相比之下 EventBus 的优点是代码简洁，使用简单，并将事件发布和订阅充分解耦。

下列源码基于EventBus3.0.0进行分析

@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface Subscribe {
  //处理事件线程
    ThreadMode threadMode() default ThreadMode.POSTING;

    /**
     * If true, delivers the most recent sticky event (posted with
     * {@link EventBus#postSticky(Object)}) to this subscriber (if event available).
     */
  //是否粘性，详见postSticky
    boolean sticky() default false;

    /** Subscriber priority to influence the order of event delivery.
     * Within the same delivery thread ({@link ThreadMode}), higher priority subscribers will receive events before
     * others with a lower priority. The default priority is 0. Note: the priority does *NOT* affect the order of
     * delivery among subscribers with different {@link ThreadMode}s! */
  //优先级
    int priority() default 0;
}

Register

public void register(Object subscriber) {
  //获取类中使用@Subscribe字段的方法列表
  //既然不是APT，不过也能理解，这部分是动态去注册的
  Class<?> subscriberClass = subscriber.getClass();
  List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
  synchronized (this) {
    for (SubscriberMethod subscriberMethod : subscriberMethods) {
      //添加到数组缓存中
      subscribe(subscriber, subscriberMethod);
    }
  }
}
//findSubscriberMethods
List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
  //METHOD_CACHE缓存，防止重新找，提高性能
  List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass);
  if (subscriberMethods != null) {
    return subscriberMethods;
  }
  //遍历方法和注解得到注册了方法列表
  //通过Class做到其所有的方法，再找到其使用了Subscribe注解的方法
  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;
  }
}

Post

public void post(Object event) {
  //获取当前线程，将message添加到消息队列中
  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;
    }
  }
}

postSingleEvent

拆分事件类型（包括事件、事件父类、实现接口）

private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
  Class<?> eventClass = event.getClass();
  boolean subscriptionFound = false;
  //事件是否继承，默认为true，也就是说会发送一个事件及其父类和接口的所以事件类
  if (eventInheritance) {
    //找到Event以及Event的父类和其所实现的接口
    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);
    }
    //如果发送一个没有被注册过的事件，会默认发送一个NoSubscriberEvent
    if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
        eventClass != SubscriberExceptionEvent.class) {
      post(new NoSubscriberEvent(this, event));
    }
  }
}

postSingleEventForEventType

找到具体的消费类，进行事件消费

private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
  //找到注册此event的事件类列表，register时候将其放进去
  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;
}

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 {
        //不在主线程发送，则通过Handler继续消费
        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);
  }
}

对ThreadMode源码分析可知：

PostThread`：默认的 ThreadMode，表示在执行 Post 操作的线程直接调用订阅者的事件响应方法，不论该线程是否为主线程（UI 线程）。当该线程为主线程时，响应方法中不能有耗时操作，否则有卡主线程的风险。适用场景：对于是否在主线程执行无要求，但若 Post 线程为主线程，不能耗时的操作；
MainThread：在主线程中执行响应方法。如果发布线程就是主线程，则直接调用订阅者的事件响应方法，否则通过主线程的 Handler 发送消息在主线程中处理——调用订阅者的事件响应函数。显然，MainThread类的方法也不能有耗时操作，以避免卡主线程。适用场景：必须在主线程执行的操作；
BackgroundThread：在后台线程中执行响应方法。如果发布线程不是主线程，则直接调用订阅者的事件响应函数，否则启动唯一的后台线程去处理。由于后台线程是唯一的，当事件超过一个的时候，它们会被放在队列中依次执行，因此该类响应方法虽然没有PostThread类和MainThread类方法对性能敏感，但最好不要有重度耗时的操作或太频繁的轻度耗时操作，以造成其他操作等待。适用场景：操作轻微耗时且不会过于频繁，即一般的耗时操作都可以放在这里；
Async：不论发布线程是否为主线程，都使用一个空闲线程来处理。和BackgroundThread不同的是，Async类的所有线程是相互独立的，因此不会出现卡线程的问题。适用场景：长耗时操作，例如网络访问。

postSticky

发送粘性事件

public void postSticky(Object event) {
  synchronized (stickyEvents) {
    //将事件缓存到stickyEvents中，当再次register时候，会去stickyEvents中获取是否存在，如已经缓存则直接消费
    //使用场景：如果您正在跟踪用户的位置，或简单的缓存数据，跟踪电池电量等，您可以使用粘性事件。
    stickyEvents.put(event.getClass(), event);
  }
  // Should be posted after it is putted, in case the subscriber wants to remove immediately
  post(event);
}
//register中调用subscribe
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
  ...略
  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);
    }
  }
}

Priority

优先级的应用：高优先级的事件处理函数可以终止事件的传递，通过cancelEventDelivery方法，但有一点需要注意，这个事件的ThreadMode必须是PostThread，并且只能终于它在处理的事件。

 private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
   //...略
     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;
     }
   }
   //...略
 }

cancelEventDelivery

终止事件分发

public void cancelEventDelivery(Object event) {
  PostingThreadState postingState = currentPostingThreadState.get();
  
  if (!postingState.isPosting) {//事件必须在发送中
    throw new EventBusException(
      "This method may only be called from inside event handling methods on the posting thread");
  } else if (event == null) {//事件必须非null
    throw new EventBusException("Event may not be null");
  } else if (postingState.event != event) {//事件必须是当前正在处理
    throw new EventBusException("Only the currently handled event may be aborted");
  } else if (postingState.subscription.subscriberMethod.threadMode != ThreadMode.POSTING) {
    //事件必须在发送线程中处理
    throw new EventBusException(" event handlers may only abort the incoming event");
  }
  //canceld标示置为true，postSingleEventForEventType中进行中断
  postingState.canceled = true;
}

注意事项

如消费事件的线程是在主线程，不宜做耗时操作
post的事件会连着其父类和所实现的接口一起分发
register&unregister一定要配对出现，否则会引发内存泄露

EventBus原理剖析

EventBus

Subscribe

Register

Post

postSingleEvent

postSingleEventForEventType

postToSubscription

postSticky

Priority

cancelEventDelivery

注意事项

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读