guava eventbus 原理+源码分析

前言:

guava提供的eventbus可以很方便的处理一对多的事件问题， 最近正好使用到了,做个小结,使用的demo网上已经很多了,不再赘述,本文主要是源码分析+使用注意点+新老版本eventbus实现方式对比

一.原理

将定义的hander注册到eventbus中，eventbus遍历该handler及其父类中含有@subscribe注解的方法，封装成subscriber对象，一个event会对应多个方法,Map<EventType.class,List<Subscriber>>,但既然是guava出品,这种情况下一定会用自己家的MultiMap了,接收到event后根据类型匹配对应的subscriber去执行,接下来从源码角度探究下

 二.源码分析

主要分析注册与分发处理,会贴相关的源码的注释(guava版本github 2021 1月版本),方便你阅读

1.注册流程

分析之前我们先简要拓展下关于guava cache的用法,compute if absent,不存在则计算,对应getOrLoad方法(暴露给用户的是get()),有则直接返回,

注册流程抓住一个关键点即可,即一个subscriber对应一个被@subscriber标记的method,为了方便阅读,我把代码贴到一起

  /** Registers all subscriber methods on the given listener object. */
  void register(Object listener) {
    // key-eventType.class value-List<Subscriber>,一个subscriber对应一个方法
    Multimap<Class<?>, Subscriber> listenerMethods = findAllSubscribers(listener);

    for (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<>();
        // eventType.class不存在时才put,concurrenthashmap的putIfAbsent()
        // 有可能为null,用newSet替换
        eventSubscribers =
            MoreObjects.firstNonNull(subscribers.putIfAbsent(eventType, newSet), newSet);
      }
      // 添加
      eventSubscribers.addAll(eventMethodsInListener);
    }
  }
  
  
  /**
   * Returns all subscribers for the given listener grouped by the type of event they subscribe to.
   */
  private Multimap<Class<?>, Subscriber> findAllSubscribers(Object listener) {
    Multimap<Class<?>, Subscriber> methodsInListener = HashMultimap.create();
    Class<?> clazz = listener.getClass();
    for (Method method : getAnnotatedMethods(clazz)) {
      Class<?>[] parameterTypes = method.getParameterTypes();
      Class<?> eventType = parameterTypes[0];
      // 创建subscriber时,如果未添加@AllowConcurrentEvents注解则生成同步的subscriber
      methodsInListener.put(eventType, Subscriber.create(bus, listener, method));
    }
    return methodsInListener;
  }

  private static ImmutableList<Method> getAnnotatedMethods(Class<?> clazz) {
    try {
      return subscriberMethodsCache.getUnchecked(clazz);
    } catch (UncheckedExecutionException e) {
      throwIfUnchecked(e.getCause());
      throw e;
    }
  }

// 映射关系缓存,getOrload
  private static final LoadingCache<Class<?>, ImmutableList<Method>> subscriberMethodsCache =
      CacheBuilder.newBuilder()
          .weakKeys()
          .build(
              new CacheLoader<Class<?>, ImmutableList<Method>>() {
                @Override
                public ImmutableList<Method> load(Class<?> concreteClass) throws Exception {
                  return getAnnotatedMethodsNotCached(concreteClass);
                }
              });

private static ImmutableList<Method> getAnnotatedMethodsNotCached(Class<?> clazz) {
    // 获得listener的所有父类及自身的class(包括接口)
    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();
          // 参数校验,@subscribe注解的方法有且有能有一个非原始类型参数
          checkArgument(
              parameterTypes.length == 1,
              "Method %s has @Subscribe annotation but has %s parameters. "
                  + "Subscriber methods must have exactly 1 parameter.",
              method,
              parameterTypes.length);

          checkArgument(
              !parameterTypes[0].isPrimitive(),
              "@Subscribe method %s's parameter is %s. "
                  + "Subscriber methods cannot accept primitives. "
                  + "Consider changing the parameter to %s.",
              method,
              parameterTypes[0].getName(),
              Primitives.wrap(parameterTypes[0]).getSimpleName());

          MethodIdentifier ident = new MethodIdentifier(method);
          // 重写的方法只放入一次
          if (!identifiers.containsKey(ident)) {
            identifiers.put(ident, method);
          }
        }
      }
    }
    return ImmutableList.copyOf(identifiers.values());
  }


  // 创建subscriber
  static Subscriber create(EventBus bus, Object listener, Method method) {
    return isDeclaredThreadSafe(method)
        ? new Subscriber(bus, listener, method)
        : new SynchronizedSubscriber(bus, listener, method);
  }

  @VisibleForTesting
  static final class SynchronizedSubscriber extends Subscriber {

    private SynchronizedSubscriber(EventBus bus, Object target, Method method) {
      super(bus, target, method);
    }

    @Override
    void invokeSubscriberMethod(Object event) throws InvocationTargetException {
      synchronized (this) {
        super.invokeSubscriberMethod(event);
      }
    }
  }

值得注意的是subscriber的生成,即便你使用了AsyncEventbus,却没有在处理方法上声明@AllowConcurrentEvents,那么在处理event时仍然是同步执行的,注册流程并发安全问题请看第三部分

2.分发流程

先看下如何获得event对应的subscriber

public void post(Object event) {
    Iterator<Subscriber> eventSubscribers = subscribers.getSubscribers(event);
    if (eventSubscribers.hasNext()) {
      // 分发,dispatcher有三种实现,ImmediateDispatcher(同步处理event,深度优先)
      // LegacyAsyncDispatcher(异步处理event)
      // PerThreadQueuedDispatcher(默认,同步调用,广度优先) 内置队列,可以保证同一线程内的event的顺序
      dispatcher.dispatch(event, eventSubscribers);
    } else if (!(event instanceof DeadEvent)) {
      // the event had no subscribers and was not itself a DeadEvent
      // 把所有没有被订阅的event包装成deadevent,用户可以自己定义处理deadevent的方法,作为兜底
      post(new DeadEvent(this, event));
    }
  }

  Iterator<Subscriber> getSubscribers(Object event) {
    //获得event的所有父类及自身的class(包括接口),从获取subscriber的流程来看,post一个event
    // 时,除了调用该event的处理方法也会调用该event父类的处理方法
    ImmutableSet<Class<?>> eventTypes = flattenHierarchy(event.getClass());

    List<Iterator<Subscriber>> subscriberIterators =
        Lists.newArrayListWithCapacity(eventTypes.size());

    for (Class<?> eventType : eventTypes) {
      CopyOnWriteArraySet<Subscriber> eventSubscribers = subscribers.get(eventType);
      if (eventSubscribers != null) {
        // eager no-copy snapshot
        subscriberIterators.add(eventSubscribers.iterator());
      }
    }
    // 类似flatmap,扁平化
    return Iterators.concat(subscriberIterators.iterator());
  }

  @VisibleForTesting
  static ImmutableSet<Class<?>> flattenHierarchy(Class<?> concreteClass) {
    try {
      return flattenHierarchyCache.getUnchecked(concreteClass);
    } catch (UncheckedExecutionException e) {
      throw Throwables.propagate(e.getCause());
    }
  }

  private static final LoadingCache<Class<?>, ImmutableSet<Class<?>>> flattenHierarchyCache =
      CacheBuilder.newBuilder()
          .weakKeys()
          .build(
              new CacheLoader<Class<?>, ImmutableSet<Class<?>>>() {
                // <Class<?>> is actually needed to compile
                @SuppressWarnings("RedundantTypeArguments")
                @Override
                public ImmutableSet<Class<?>> load(Class<?> concreteClass) {
                  return ImmutableSet.<Class<?>>copyOf(
                      TypeToken.of(concreteClass).getTypes().rawTypes());
                }
              });

从代码可以看出,先对该event查询上级,最后把所有event对应的subscriber返回,因此触发一个event时,其父event的subscriber也会被调用

接下来看下post,流程eventbus有三种dispatcher(ImmediaDispatcher,PerThreadDispatcher,LegacyAsyncDispatcher)eventbus使用的是PerThreadDispatcher,AsyncEventBus使用LegacyAsyncDispatcher

①ImmediaDispatcher

从名字中的Immedia"即时"就能看出这个dispatcher收到event后会立即处理,不会进行异步处理

代码如下:

 从图中可以看出ImmediaDispatcher是针对每个event,调用其全部的subscriber进行处理,即尽可能多的调用subscriber,所以是广度优先,这个dispatcher目前未被使用,了解即可

 ②PerThreadQueueDispatcher(默认的dispatcher)

同样从名称可以看出这种dispatcher是一个thread一个queue,那我们可以猜测内部有可能用了ThreadLocal,既然用了队列,说明想要起到一个缓冲event处理的过程

队列的缓冲功能使得dispatcher有能力吞吐更高的event,因此是一种深度优先策略,此外每线程每队列的方式保证了event处理过程是对于每个线程而言是有序的,同样是广度优先,对

每一个event都分发到相关的subscriber进行处理,除此之外还有一个值得称道的点,即Dispatching变量的使用,规避了递归产生的死循环问题

private static final class PerThreadQueuedDispatcher extends Dispatcher {

    // This dispatcher matches the original dispatch behavior of EventBus.

    /** 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);
      // 如果只从代码来看,PerThreadQueuedDispatcher的dispatch方法始终
      // 是单线程调用,并不需要ThreadLocal,但从拓展的角度看,当用户自定义xxeventbus自己实现分发逻辑时,PerThreadQueuedDispatcher实现了线程安全的dispatch
      //因为eventbus有可能会被多个线程调用,从框架的角度看,无论用户是否多线程调用,都应该要保证线程安全
      // 引用issue 3530中 https://github.com/google/guava/issues/3530 的一个回答 if multiple threads are dispatching to this dispatcher, they will read different values for queueForThread and dispatching.
      Queue<Event> queueForThread = queue.get();
      queueForThread.offer(new Event(event, subscribers));

      // 如果未开始分发事件则进行处理,解决subscriber递归调用post产生的死循环
      if (!dispatching.get()) {
        dispatching.set(true);
        try {
          Event nextEvent;
          // 对每一个event,分发到相关的subscribers中
          while ((nextEvent = queueForThread.poll()) != null) {
            while (nextEvent.subscribers.hasNext()) {
              nextEvent.subscribers.next().dispatchEvent(nextEvent.event);
            }
          }
        } finally {
          dispatching.remove();
          queue.remove();
        }
      }
    }

接下来看下刚刚说的dispatching的妙用demo

在guava-test下建立一个新的目录方便我们修改源码后进行测试,测试代码如下

Listener

/**
 * @author tele
 * @Description
 * @create 2020-11-23
 */
public class Listener {

    private final EventBus eventBus;

    public Listener(EventBus eventBus) {
        this.eventBus = eventBus;
    }

    @Subscribe
    public void record(String s) {
        eventBus.post(s);
        System.out.println("receive:"+ s);
    }
}

Producer

/**
 * @author tele
 * @Description
 * @create 2020-11-23
 */
public class Producer {

    public String produce() {
        return "hello";
    }
}

Main

/**
 * @author tele
 * @Description
 * @create 2020-11-23
 */
public class Main {

    public static void main(String[] args) {
        EventBus eventBus = new EventBus();
        Listener listener = new Listener(eventBus);
        Producer producer = new Producer();
        eventBus.register(listener);
        String produce = producer.produce();
        eventBus.post(produce);
    }

}

代码很简单,问题在于Listener递归调用了post方法,按照代码示意运行后会栈溢出(队列中event堆积),receive:hello永远不会打印,可事实真的如此吗?

 很奇怪是吗,并没有产生堆栈溢出的问题,反而是不停的输出receive:hello,接下来我们修改下PerThreadDispatcher的代码,将dispatching变量注释掉

  

再执行下demo

 果然溢出了,关键点就在于dispatching变量对于同一线程的递归分发进行了处理,已经处理过就不再次进行分发,这样我们的递归调用不停的产生的event得以被处理

 ③LegacyAsyncDispatcher

看名字挺奇怪的,但有async字样,所以是异步的dispatcher,LegacyAsyncDispacther是AsyncEventBus的专用dispatcher,由于将event对应的subscriber拆分后入队,多线程情况下无法保证event入队顺序,也就无法保证subscriber的调用顺序,但这样处理实现了深度优先,即尽可能多的调用不同的event的subscriber,与PerThreadDispatcher相比代码难度小了不少,由于AsyncEventBus的初始化需要传入线程池参数,所以AsyncEventBus实现了真正的异步处理

/** Implementation of a {@link #legacyAsync()} dispatcher. */
  private static final class LegacyAsyncDispatcher extends Dispatcher {

    // This dispatcher matches the original dispatch behavior of AsyncEventBus.
    //
    // We can't really make any guarantees about the overall dispatch order for this dispatcher in
    // a multithreaded environment for a couple reasons:
    //
    // 1. Subscribers to events posted on different threads can be interleaved with each other
    //    freely. (A event on one thread, B event on another could yield any of
    //    [a1, a2, a3, b1, b2], [a1, b2, a2, a3, b2], [a1, b2, b3, a2, a3], etc.)
    // 2. It's possible for subscribers to actually be dispatched to in a different order than they
    //    were added to the queue. It's easily possible for one thread to take the head of the
    //    queue, immediately followed by another thread taking the next element in the queue. That
    //    second thread can then dispatch to the subscriber it took before the first thread does.
    //
    // All this makes me really wonder if there's any value in queueing here at all. A dispatcher
    // that simply loops through the subscribers and dispatches the event to each would actually
    // probably provide a stronger order guarantee, though that order would obviously be different
    // in some cases.

    /** Global event queue. */
    private final ConcurrentLinkedQueue<EventWithSubscriber> queue =
        Queues.newConcurrentLinkedQueue();

    @Override
    void dispatch(Object event, Iterator<Subscriber> subscribers) {
      checkNotNull(event);
      // 拆分后入队
      while (subscribers.hasNext()) {
        queue.add(new EventWithSubscriber(event, subscribers.next()));
      }

      EventWithSubscriber e;
      while ((e = queue.poll()) != null) {
        e.subscriber.dispatchEvent(e.event);
      }
    }

    private static final class EventWithSubscriber {
      private final Object event;
      private final Subscriber subscriber;

      private EventWithSubscriber(Object event, Subscriber subscriber) {
        this.event = event;
        this.subscriber = subscriber;
      }
    }
  }

注意点:

1.eventbus默认使用的线程池MoreExecutors.directExecutor(),其execute方法是直接调用传入的runnable的run方法,是非异步的 

2.使用AsyncEventBus时,请在对应的方法上添加@AllowConcurrenEvents

三.从并发安全的角度出发,对比下新老版本的注册流程

本部分为补充内容,重点探讨新老版本的注册并发安全问题,可略过

从20.0开始,event bus的注册程变成了上面分析的,那么之前的版本是如何实现的呢,一起来分析下.先切到16.0 的tag,注册代码如下

显然是使用了读写锁,不加锁,eventType会相互覆盖(HashMultiMap是非线程安全的),先给eventbus加个getSubscriberByType(),记得修改下EventSubscriber的修饰符为public,然后做个多线程的测试

/**
 * @author tele
 * @Description
 * @create 2021-01-24
 */
public class ListenerA {

    @Subscribe
    public void handle(String msg) {
        System.out.println("ListenerA:" + msg);
    }

}

/**
 * @author tele
 * @Description
 * @create 2021-01-24
 */
public class ListenerB {

    @Subscribe
    public void handle(String msg) {
        System.out.println("ListenerB:" + msg);
    }

}

/**
 * @author tele
 * @Description
 * @create 2021-01-24
 */
public class Main {


    public static void main(String[] args) throws InterruptedException {

        final EventBus eventBus = new EventBus();
        final ListenerA a = new ListenerA();
        ListenerB b = new ListenerB();
        CountDownLatch countDownLatch = new CountDownLatch(6);

        Runnable r1 = ()-> {
            eventBus.register(a);
            countDownLatch.countDown();
        };
        Thread t1 = new Thread(r1);
        Thread t2 = new Thread(r1);
        Thread t3 = new Thread(r1);

        Runnable r2 = ()-> {
            eventBus.register(b);
            countDownLatch.countDown();
        };
        Thread t4 = new Thread(r2);
        Thread t5 = new Thread(r2);
        Thread t6 = new Thread(r2);

        t1.start();
        t2.start();
        t3.start();
        t4.start();
        t5.start();
        t6.start();
        countDownLatch.await();
        SetMultimap<Class<?>, EventSubscriber> subscribersByType = eventBus.getSubscribersByType();
        subscribersByType.asMap().forEach((k,v)-> {
            System.out.println("key:" + k);
            v.forEach(System.out::println);
        });
    }
}

输出结果如下:

 ok,没啥问题,接下来再修改下源码把使用读写锁的两行代码注释掉,再执行下代码

 

 输出结果如下:

显然,ListenerA的注册结果被覆盖了,这里简要说下原因,subscribersByType,k-v结构简略表示为 K-event.class ,value-Set<Listener.class>,我们知道java中的hashset不重复的特性是基于hashmap实现的.同样的,这里的SetMultiMap实际是用的HashMultiMap,翻翻源码就知道了,内部存储数据的容器是hashmap,那么这个问题就转换成了hashmap的线程安全问题了,hashmap多线程put hash相同的元素会产生丢失问题,多线程下同时put get有可能导致get 出null.了解到这我们就知道为什么要加锁了,使用读写锁的版本一直持续到19.0,从20.0开始从开始使用并发容器代替读写锁,因为对于eventbus而言始终是读远大于写,基于cow机制实现的CopyOnWriteArrayList在读写同时进行时通过延迟更新的策略不阻塞线程,对于event的处理 而言是可以接受的,因为本次event在post时没有分发到对应的subsriber,下次同类型的event触发就ok了,事实上,这种场景极少,因为从使用经历来看,一般是项目启动时就注册,分发都是需要处理逻辑时才会触发,不阻塞与每次都需要加解读锁相比,显然不阻塞的性能更好了.老版本的分发流程不再赘述,因为确实没啥好分析的了,如果你能看懂上面分析的新版本的dispatcher,当你看老版本的时候就会感觉很简单了

四.优势与缺陷

1.进程内使用,无法实现跨进程处理,需要跨进程传递消息,还是老老实实的用消息队列吧

2.和redis一样基于内存,天然的不可靠,redis好歹还有aof和rdb,可event bus没有任何持久化机制

3.个人对新版的Subscriber实现方式有点看法,没必须要把线程池参数传递给Subscriber,因为Subscriber只是被执行者,16.0的版本线程池参数是AsyncEventBus持有

4.优势:简单,开箱即用

五.小结

1.只分析了注册与分发流程,异常处理之类的没有涉及,用法的话,网上已经很多了,不再赘述

2.event bus的代码很巧妙,细细品味还有很多巧妙之处,比如上面那个dispatching变量

六.参考文档

1.github https://github.com/google/guava/wiki/EventBusExplained#for-producers