Guava-EventBus

前言

设计模式中有一种模式叫观察者模式，观察者模式能解耦方法与方法之间的关系。这里使用《Head First 设计模式》中的例子来说。


image.png

image.png
image.png

image.png


可以看出 观察者和订阅者是一个松耦合的关系。


主题对象维护了一个观察者列表，当有事件更新时，主题对象会遍历观察者列表，并依次执行通知操作。


EventBus简介

eventbus就是对上面观察者模式的一个比较优雅的实现。它没有采用上面介绍的这种定义主题接口和观察者接口的模式，而是定义了一个 @Subscribe 注解，将它加在你方法上。就表示如果有事件发生，这个方法会接受事件参数并执行。如下代码：

//start DataObjectOne.java =====================
import com.google.common.eventbus.Subscribe;

public class DataObjectOne {
    @Subscribe
    public void func(String msg){
        System.out.println("String msg:"+msg);
    }
    
    public void test(String msg){
        System.out.println("String test msg:"+msg);
    }
}
//end DataObjectOne.java =====================

//start DataObjectTwo.java =====================
import com.google.common.eventbus.Subscribe;

public class DataObjectTwo {
    @Subscribe
    public void func(Integer msg){
        System.out.println("Integer msg:"+msg);
    }
}
//end DataObjectOne.java =====================


//start EventBusTest.java =====================
import com.google.common.eventbus.EventBus;

public class EventBusTest {
    public static void main(String[] args) {
        EventBus eventBus=new EventBus();
        DataObjectOne one=new DataObjectOne();
        DataObjectTwo two=new DataObjectTwo();
        eventBus.register(one);
        eventBus.register(two);
        eventBus.post("test");
        eventBus.post(123);
    }
}
//end EventBusTest.java =====================

//输出结果：
String msg:test
Integer msg:123

实际上EventBus的原理不算很麻烦，
1、首先它会在注册的时候，通过反射将被注册类中所有的 有 @Subscribe 注解的方法以k-v的形式(k是方法的参数类型(Type),v是Method对象) 存储在Multimap里面(就当做一个map就好)。
2、然后在post的时候，会通过反射获取post方法参数的Type，然后再去Multimap里面比对，获取Method。
3、获取到Method对象后，通过执行器去执行(实际上就是另外起了一个线程通过反射去执行Method)。


下面将进行代码的流程梳理。

EventBus代码梳理

从上面的测试例子可以知道代码梳理有三个入口，第一个是EventBus的初始化，一个是register，一个是post。先来看看初始化。


EventBus初始化

EventBus初始化的核心方法如下：

  /**
   * Creates a new EventBus with the given {@code identifier}.
   *
   * @param identifier a brief name for this bus, for logging purposes. Should be a valid Java
   *     identifier.
   */
  public EventBus(String identifier) {
    this(
        identifier,  //eventbus的一个简要名字
        MoreExecutors.directExecutor(), //创建执行器
        Dispatcher.perThreadDispatchQueue(), //创建调度器
        LoggingHandler.INSTANCE); //日志部分
  }

执行器的创建：

public static Executor directExecutor() {
    return DirectExecutor.INSTANCE;
  }

@GwtCompatible
enum DirectExecutor implements Executor {
  INSTANCE;

  @Override
  public void execute(Runnable command) {
    command.run(); //执行器接受的参数是一个Runnable，可见这是执行一个线程
  }

  @Override
  public String toString() {
    return "MoreExecutors.directExecutor()";
  }
}

另外这个directExecutor()的注释里面有写到：

This executor is appropriate for tasks that are lightweight and not deeply chained. Inappropriate directExecutor usage can cause problems, and these problems can be difficult to reproduce because they depend on timing.


执行器适用于轻量级的，调用链不太深的任务。乱用directExecutor会导致一些难以重现和排查的问题。
总的来说就是执行器不能执行太复杂的、太耗时和太重量级的任务，否则会导致程序进程停滞等。


调度器：调度器会给依次给排队的订阅者分派事件，确保发布的所有事件都被分配给了各自的订阅者，且顺序执行。当所有的订阅者被分派到了事件后，会创建一个宽度优先的调度顺序程序，例如会先执行A事件的所有订阅者的任务，然后在执行B时间的任务。表示任务的执行是有序的。 ```java static Dispatcher perThreadDispatchQueue() { return new PerThreadQueuedDispatcher(); } ``` 
  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);
      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();
        }
      }
    }

    private static final class Event {
      private final Object event;
      private final Iterator<Subscriber> subscribers;

      private Event(Object event, Iterator<Subscriber> subscribers) {
        this.event = event;
        this.subscribers = subscribers;
      }
    }
  }

Register 注册事件

  public void register(Object object) {
    subscribers.register(object); 
  }

  void register(Object listener) {
    Multimap<Class<?>, Subscriber> listenerMethods = findAllSubscribers(listener);//获取监听方法 

    for (Entry<Class<?>, Collection<Subscriber>> entry : listenerMethods.asMap().entrySet()) {
        //下面的内容就是做了一些包装，将订阅者存放到了一个线程安全的set中。
      Class<?> eventType = entry.getKey();
      Collection<Subscriber> eventMethodsInListener = entry.getValue();

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

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

      eventSubscribers.addAll(eventMethodsInListener);
    }
  }

  private Multimap<Class<?>, Subscriber> findAllSubscribers(Object listener) {
    Multimap<Class<?>, Subscriber> methodsInListener = HashMultimap.create(); //创建Multimap
    Class<?> clazz = listener.getClass(); 
    for (Method method : getAnnotatedMethods(clazz)) { //遍历有 Subscribe注解的方法 
      Class<?>[] parameterTypes = method.getParameterTypes(); //获取方法的参数类型
      Class<?> eventType = parameterTypes[0]; 
      methodsInListener.put(eventType, Subscriber.create(bus, listener, method)); //存放到map中 
    }
    return methodsInListener;
  }

getAnnotatedMethods

  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> getAnnotatedMethods(Class<?> clazz) {
    try {
      return subscriberMethodsCache.getUnchecked(clazz); //这个从一个cache里面读取它有订阅注解的方法
    } catch (UncheckedExecutionException e) {
      throwIfUnchecked(e.getCause());
      throw e;
    }
  }

  //创建cache的时候就去遍历method
  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()) {
          
          //这里就可以看到它会判断method是否有Subscribe 注解
          
        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);

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

post方法

  public void post(Object event) {
    Iterator<Subscriber> eventSubscribers = subscribers.getSubscribers(event);//获取事件的订阅对象 核心方法之一
    if (eventSubscribers.hasNext()) {
      dispatcher.dispatch(event, eventSubscribers); //往调度器里面调度event 核心方法
    } else if (!(event instanceof DeadEvent)) {
      // the event had no subscribers and was not itself a DeadEvent
      post(new DeadEvent(this, event));
    }
  }

  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);
      Queue<Event> queueForThread = queue.get(); 
      queueForThread.offer(new Event(event, subscribers));//往调度queue里面添加

      if (!dispatching.get()) {  //保证线程安全的
        dispatching.set(true);
        try {
          Event nextEvent;
          while ((nextEvent = queueForThread.poll()) != null) { //取出Event
            while (nextEvent.subscribers.hasNext()) {
              nextEvent.subscribers.next().dispatchEvent(nextEvent.event); //调度event 核心方法
            }
          }
        } finally {
          dispatching.remove();
          queue.remove();
        }
      }
    }

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

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

总结

如上代码梳理也验证了EventBus的原理：
1、首先它会在注册的时候，通过反射将被注册类中所有的 有 @Subscribe 注解的方法以k-v的形式(k是方法的参数类型(Type),v是Method对象) 存储在Multimap里面(就当做一个map就好)。
2、然后在post的时候，会通过反射获取post方法参数的Type，然后再去Multimap里面比对，获取Method。
3、获取到Method对象后，通过执行器去执行(实际上就是起了一个线程通过反射去执行Method)。


值得借鉴的点

1 为了防止重入dispatch，使用了 final ThreadLocal变量。

 private final ThreadLocal<Boolean> dispatching =
        new ThreadLocal<Boolean>() {
          @Override
          protected Boolean initialValue() {
            return false;
          }
        };

。。。略了部分内容

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

2 数据结构：
ConcurrentMap：一个线程安全的，有原子性保证的map. 是一个接口
CopyOnWriteArraySet：是由CopyOnWriteArrayList实现的结构，可以理解为线程安全的HashSet,但是底层是动态数组实现的。适用于查询多，修改少。修改时需要复制整个数组，开销大。

参考内容：
Head First 设计模式
https://blog.csdn.net/wuyuxing24/article/details/95505102
https://www.cnblogs.com/dennyzhangdd/p/9324483.html

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/junjiedeng/p/14611595.html