EventBus源码学习笔记(二)

EventBus深入学习二

开始研究源码的设计思路,从Listener注册出发,EventBus 如何维护监听者信息,到Publisher发送消息,消息以怎样的渠道分发给所有的Listener, 顺序如何保证,传递性如何保证,出现异常如何处理,找不到监听者怎么处理等等

EventBus

这个类相当于一个中转站,Publisher 调用它的 post(Object) 来推送事件;然后将事件一次推送给注册的Listener

1. 注册关系的维护

在初始化s时, EventBus对象会维护一个 private final SubscriberRegistry subscribers = new SubscriberRegistry(this); 实例, 这个就是维护订阅关系的核心类

注册方法如下

 /**
   * Registers all subscriber methods on {@code object} to receive events.
   *
   * @param object  object whose subscriber methods should be registered.
   */
  public void register(Object object) {
    subscribers.register(object);
  }

接着我们看下这个类的具体实现

SubscriberRegistry.java

/**
   * All registered subscribers, indexed by event type.
   *
   * <p>The {@link CopyOnWriteArraySet} values make it easy and relatively lightweight to get an
   * immutable snapshot of all current subscribers to an event without any locking.
   */
  private final ConcurrentMap<Class<?>, CopyOnWriteArraySet<Subscriber>> subscribers =
      Maps.newConcurrentMap();
      

  /**
   * Registers all subscriber methods on the given listener object.
   */
  void register(Object listener) {
    Multimap<Class<?>, Subscriber> listenerMethods = findAllSubscribers(listener);

    for (Map.Entry<Class<?>, Collection<Subscriber>> entry : listenerMethods.asMap().entrySet()) {
      Class<?> eventType = entry.getKey();
      Collection<Subscriber> eventMethodsInListener = entry.getValue();

      CopyOnWriteArraySet<Subscriber> eventSubscribers = subscribers.get(eventType);

      if (eventSubscribers == null) {
        CopyOnWriteArraySet<Subscriber> newSet = new CopyOnWriteArraySet<Subscriber>();
        eventSubscribers = MoreObjects.firstNonNull(
            subscribers.putIfAbsent(eventType, newSet), newSet);
      }

      eventSubscribers.addAll(eventMethodsInListener);
    }
  }

subscribers : - 对象初始化时创建 - 维护的是EventType -> Listener的映射关系,value为一个集合,说明一个事件可以推送给多个Listener - 监听者,可以有可以监听多个不同类型的事件

注册流程: - 根据注册的对象,将其中所有的回调方法都捞出来 - 将上步的结果塞入 subscribers 集合中; key为 Listener的类名

注册流程详解

注册目的就是发布消息后, EventBus 可以将这个Event传递” Listener(即订阅方)

为了实现上面的目的,如果要我们自己实现,会怎么做?

  • 将类中,所有包含@Subscribe 注解的方法捞出来
  • 方法的第一个参数就是 Event, 因为注册的目的是为了实现回调, 所以封装一个类,包含这个Listener对象的引用 + 要执行的方法

上面注册的实际实现和上面的步骤差不多

  1. 获取所有包含注解的方法
    • 实际的代码如下
private static ImmutableList<Method> getAnnotatedMethodsNotCached(Class<?> clazz) {
Set<? extends Class<?>> supertypes = TypeToken.of(clazz).getTypes().rawTypes();
Map<MethodIdentifier, Method> identifiers = Maps.newHashMap();
for (Class<?> supertype : supertypes) {
  for (Method method : supertype.getDeclaredMethods()) {
    if (method.isAnnotationPresent(Subscribe.class) && !method.isSynthetic()) {
      // TODO(cgdecker): Should check for a generic parameter type and error out
      Class<?>[] parameterTypes = method.getParameterTypes();
      checkArgument(parameterTypes.length == 1,
          "Method %s has @Subscribe annotation but has %s parameters."
              + "Subscriber methods must have exactly 1 parameter.",
          method, parameterTypes.length);

      MethodIdentifier ident = new MethodIdentifier(method);
      if (!identifiers.containsKey(ident)) {
        identifiers.put(ident, method);
      }
    }
  }
}
return ImmutableList.copyOf(identifiers.values());
}
  • 看下上面的实现,非常有意思的是,不仅将改对象中的所有@Subscribe注解的方法捞出来,连父类中的也不放过;就是这个TypeToken.of(clazz).getTypes().rawTypes();
  • 从上面的限定,也可以看出,对于回调方法是有要求的: 有且只能有一个参数, checkArgument(parameterTypes.length == 1,xxx)
  • 过滤重载的回调方法(这点比较有意思,搞了个Map, key由方法名+方法参数确认(MethodIdentifier的equals方法重写了), 而不是直接用集合的contains方法, 请注意其中的区别)
  • method.isAnnotationPresent(Subscribe.class) && !method.isSynthetic() 这个判断条件的后一个可以参考http://www.xue163.com/2122/1/21224778.html

  1. 将上面的方法转换为Map, 看这个 Multimap<Class<?>, Subscriber> listenerMethods = findAllSubscribers(listener);
    • key为事件类型Event.class; value为一个包含 Listener, Method, EventBus 实例的对象 Subscriber

  1. 将上面的map塞入subscribers 集合
    • subscribers 集合包含的是所有的 (事件 -> 监听者回调集合)Event -> Set<Listener.Method>
    • 简单的迭代即可实现塞数据了
    • 根据 subscribers 的数据结构,其实可以看到,一个Listener对象,如果注册多次,最终的效果其实是一样的,这个监听者,并不会被调用多次; 如果一个Lisntener类,有多个对象,则注册后,每个对象的回调都会执行到;
      • 实现原因: Set 集合存储 Subscriber对象
      • Subscriber 的 hashcode & equals 方法没有重写

到此, 注册完毕;注销的方法和上面差不多,唯一的区别是最后一个是向 subscribers 塞数据,一个是从其中删数据而已

题外话

   如果我们想获取工程中所有包含某个注解的类可以怎么办?
   
   - 如果是用spring的话, 可以考虑  `ApplicationContext.getBeansWithAnnotation()`
   
   获取工程中,所有包含某个注解的方法,除了上面的主动注册,有什么其他的方法?

###2. 推送事件

发布方,调用EventBus.post(Object) 方法实现消息的推送

预测

正式开始之前,我们可以先预测一下,当发布方调用了这个方法之后,会执行那些action

  1. 根据事件类型,可以从注册关系表subscribers中获取出所有的监听者,以及对应的回调方法, 放在一个集合中
  2. 因为监听的先后顺序可能有要求,那么将上面的集合进行排序
  3. 循环遍历上面,依次执行

上面是正向的操作流程,接着一些异常情况和边界也需要考虑下

  1. 如果一个事件找不到订阅者如何处理
  2. 如果某个监听者执行完毕之后,希望其之后的监听者都不能接受这个事件(类似web应用中的拦截器,如果被拦截了,如果被拦截了,后面的拦截器也没必要继续执行)
  3. 某个监听者很遗憾的抛出了个异常,会不会整调链路都挂掉

深入解读

带着上面的臆测,来实际看下EventBus自己是怎么玩的

 /**
   * 将事件推送给所有的监听者,不管监听者是否正常处理,都是正确返回
   * Posts an event to all registered subscribers.  This method will return
   * successfully after the event has been posted to all subscribers, and
   * regardless of any exceptions thrown by subscribers.
   *
   * 如果一个事件没有监听者,且该事件不是 DeadEvent, 则转为 DeadEvent并重新推送
   * <p>If no subscribers have been subscribed for {@code event}'s class, and
   * {@code event} is not already a {@link DeadEvent}, it will be wrapped in a
   * DeadEvent and reposted.
   *
   * @param event  event to post.
   */
  public void post(Object event) {
    Iterator<Subscriber> eventSubscribers = subscribers.getSubscribers(event);
    if (eventSubscribers.hasNext()) {
      dispatcher.dispatch(event, eventSubscribers);
    } else if (!(event instanceof DeadEvent)) {
      // the event had no subscribers and was not itself a DeadEvent
      post(new DeadEvent(this, event));
    }
  }

上面的解释比较清楚, 基本上核心的推送就是 dispatcher.dispatch(event, eventSubscribers);

实际的使用的是 PerThreadQueuedDispatcher 推送代码如下,逻辑比较清晰,将Event塞入队列, 然后将队列中的所有消息依次推送给所有的订阅者

  • 可以参考下面的设计,优雅的避免重复推送的问题
  • 考虑为什么要先写入队列,然后再依次推送队列中的事件
/**
* Per-thread queue of events to dispatch.
*/
private final ThreadLocal<Queue<Event>> queue =
   new ThreadLocal<Queue<Event>>() {
     @Override
     protected Queue<Event> initialValue() {
       return Queues.newArrayDeque();
     }
   };

/**
* Per-thread dispatch state, used to avoid reentrant event dispatching.
*/
private final ThreadLocal<Boolean> dispatching =
   new ThreadLocal<Boolean>() {
     @Override
     protected Boolean initialValue() {
       return false;
     }
   };

@Override
void dispatch(Object event, Iterator<Subscriber> subscribers) {
 checkNotNull(event);
 checkNotNull(subscribers);
 Queue<Event> queueForThread = queue.get();
 queueForThread.offer(new Event(event, subscribers));

 if (!dispatching.get()) {
   dispatching.set(true);
   try {
     Event nextEvent;
     while ((nextEvent = queueForThread.poll()) != null) {
       while (nextEvent.subscribers.hasNext()) {
         nextEvent.subscribers.next().dispatchEvent(nextEvent.event);
       }
     }
   } finally {
     dispatching.remove();
     queue.remove();
   }
 }
}

最终真正执行推送Event的是这个方法 com.google.common.eventbus.Subscriber#dispatchEvent

/**
   * Dispatches {@code event} to this subscriber using the proper executor.
   */
  final void dispatchEvent(final Object event) {
    executor.execute(new Runnable() {
      @Override
      public void run() {
        try {
          invokeSubscriberMethod(event);
        } catch (InvocationTargetException e) {
          bus.handleSubscriberException(e.getCause(), context(event));
        }
      }
    });
  }
  
  /**
   * Invokes the subscriber method. This method can be overridden to make the invocation
   * synchronized.
   */
  @VisibleForTesting
  void invokeSubscriberMethod(Object event) throws InvocationTargetException {
    try {
      method.invoke(target, checkNotNull(event));
    } catch (IllegalArgumentException e) {
      throw new Error("Method rejected target/argument: " + event, e);
    } catch (IllegalAccessException e) {
      throw new Error("Method became inaccessible: " + event, e);
    } catch (InvocationTargetException e) {
      if (e.getCause() instanceof Error) {
        throw (Error) e.getCause();
      }
      throw e;
    }
  }

3. 图解

上面从源码的角度,对整个流程顺了一遍,下面的图对几个主要的类结构进行了抽取,并对上面的几个方法进行了简要的说明

图一, 将上面说明的几个类属性 + 方法进行了说明

图二, 对逻辑进行列举

4. 新技能

1.根据class,获取所有超类集合 (EventBus的实际使用中,Event的超类集合都塞入了缓存,加快查询速度)

TypeToken.of(concreteClass).getTypes().rawTypes());

2.获取类中标有注解的方法

private static ImmutableList<Method> getAnnotatedMethodsNotCached(Class<?> clazz, Class annotationClz) {
    Set<? extends Class<?>> supertypes = TypeToken.of(clazz).getTypes().rawTypes();
    Map<MethodIdentifier, Method> identifiers = Maps.newHashMap();
    for (Class<?> supertype : supertypes) {
      for (Method method : supertype.getDeclaredMethods()) {
        if (method.isAnnotationPresent(annotationClz) && !method.isSynthetic()) {
          // TODO(cgdecker): Should check for a generic parameter type and error out
          Class<?>[] parameterTypes = method.getParameterTypes();
          
          MethodIdentifier ident = new MethodIdentifier(method);
          if (!identifiers.containsKey(ident)) {
            identifiers.put(ident, method);
          }
        }
      }
    }
    return ImmutableList.copyOf(identifiers.values());
  }
  
  private static final class MethodIdentifier {

    private final String name;
    private final List<Class<?>> parameterTypes;

    MethodIdentifier(Method method) {
      this.name = method.getName();
      this.parameterTypes = Arrays.asList(method.getParameterTypes());
    }

    @Override
    public int hashCode() {
      return Objects.hashCode(name, parameterTypes);
    }

    @Override
    public boolean equals(@Nullable Object o) {
      if (o instanceof MethodIdentifier) {
        MethodIdentifier ident = (MethodIdentifier) o;
        return name.equals(ident.name) && parameterTypes.equals(ident.parameterTypes);
      }
      return false;
    }
  }

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏数据结构与算法

洛谷P2792 [JSOI2008]小店购物(最小树形图)

一开始的思路:新建一个虚点向每个点连边,再加上题面中给出的边,边权均为大小*需要购买的数量

901
来自专栏大内老A

采用一个自创的"验证框架"实现对数据实体的验证[编程篇]

昨天晚上突发奇想,弄了一个简易版的验证框架,用于进行数据实体的验证。目前仅仅实现基于属性的声明式的验证,即通过自定义特性(Custom Attribute)的方...

2596
来自专栏数据结构与算法

BZOJ1269: [AHOI2006]文本编辑器editor

Descriptio 这些日子,可可不和卡卡一起玩了,原来可可正废寝忘食的想做一个简单而高效的文本编辑器。你能帮助他吗? 为了明确任务目标,可可对“文本编辑器...

2757
来自专栏java学习

Java每日一练(2017/6/8)

Java基础 | 数据库 | Android | 学习视频 | 学习资料下载 课前导读 ●回复"每日一练"获取以前的题目! ●答案公布时间:为每期发布题目的第二...

2707
来自专栏三好码农的三亩自留地

Java动态代理-实战

说动态代理之前,要先搞明白什么是代理,代理的字面意思已经很容易理解了,我们这里撇开其他的解释,我们只谈设计模式中的代理模式

2402
来自专栏我的小碗汤

golang 设置 http response 响应头与坑

之前遇到个问题,在一段代码中这样设置WriteHeader,最后在header中取Name时怎么也取不到。

852
来自专栏Android 研究

OKHttp源码解析(五)--OKIO简介及FileSystem

okio是由square公司开发的,它补充了java.io和java.nio的不足,以便能够更加方便,快速的访问、存储和处理你的数据。OKHttp底层也是用该库...

1612
来自专栏jeremy的技术点滴

JVM的Finalization Delay引起的OOM

3808
来自专栏学习力

《Java从入门到放弃》JavaSE入门篇:练习——单身狗租赁系统

3774
来自专栏butterfly100

ConcurrentHashMap源码阅读

1. 前言 HashMap是非线程安全的,在多线程访问时没有同步机制,并发场景下put操作可能导致同一数组下的链表形成闭环,get时候出现死循环,导致CPU利用...

2997

扫码关注云+社区