guava之EventBus

观察者模式

这个讲的地方特别多，随便百度就好，这里随意贴了一个别人的链接

观察者模式（Observer模式）详解

解决的问题

观察者模式解决的场景就是：解耦。将一些和主业务逻辑不强相关的逻辑可以解耦出来，统一处理。

Eventbus

EventBus提供了两种观察方式：同步和异步

• 同步方式：post()和对应的事件处理器EventHandler逻辑在同一个线程执行。
• 异步方式：事件处理器EventHandler逻辑在指定线程池中执行，和post()事件的线程没有关系。

基本设计：

主要的类：

• private final HandlerFindingStrategy finder：这个是用于解析时间处理器的，EventBus只是实现了使用@Subscribe注解来绑定事件和其对应处理器的的：AnnotatedHandlerFinder
• SetMultimap<Class<?>, EventHandler> handlersByType：这个就是缓存的finder的一个结果，保存了事件及其对应的处理器的关系。它是一个map结构：key=事件类型(即@Subscribe注解方法的入参的类型)，value=Set<EventHandler>,即使这个事件对应的处理器。EventHandler实际封装的就是@Subscribe注解方法的反射Method对象。ps：实际EventBus中value不是个set，而是guava自己实现的SetMultiMap结构。
• ThreadLocal<Queue<EventWithHandler>> eventsToDispatch：这个地方分两点：一个是之类为啥用ThreadLocal，第二是有了handlersByType，为啥这里还需要一个Queue。
  • 先说为啥有了handlersByType，还需要这个queue。因为不需要这个queue也可以实现的，方法就是：在post()事件的时候，根据事件类型从handlersByType获得这个事件所有的EventHandler，然后分别调用这些EventHandler就好了。但是这里并没有这么做，而是在post()的时候，根据事件类型从handlersByType获得这个事件的所有EventHandler，但是不是分别调用这个EventHandler，而是遍历这些EventHandler，封装成EventWithHandler，然后入队到Queue<EventWithHandler>，然后再开始分派：不断从queue中出队，然后调用EventWithHandler中封装的EventHandler，入参就是EventWithHandler的Event
  • 为什么是ThreadLocal？EventBus是同步的，即post()事件和EventHandler的处理是在同一个线程中执行。比如我初始化了一个eventBus，然后线程A调用eventBus.post(EventA),然后线程B也用这个eventBus.post(事件B)，那么事件A和事件B的Handler逻辑应该在线程A和线程B分别执行。如果不用ThreadLocal，而是一个多线程共享的Queue，要实现post和Handler 在同一个线层中执行，是不是还挺麻烦的，需要记录post是哪个线程执行的，然后dispatcher分发的时候将事件分发给对应的线程，但是使用ThreadLocal，这个对应关系ThreadLocal搞定了。 而且ThreadLocal是线程不共享的，也就实现了多线程post事件，操作这个Queue是线程安全的。

  • AsyncEentBus中的Queue<EventWithHandler>是ConcurrentLinkedQueue<EventWithHandler>，这又是为啥
    首先AsyncEventBus post事件和该事件的处理器Handler是不同的线程，且是哪个线程根本就不重要。所以会直接将这个事件处理器扔给线程池就好，利用线程池的调度来找到线程去驱动这个事件Handler的逻辑的执行。
    但是这里会保证，先post事件，先提交给EventBus的事件，会先提交给线程，但是谁先执行就不好说了，这个看线程池的调度。另外就是多线程post事件操作Queue是需要线程安全的，所以这里使用的是线程安全的容器。

• handlersByTypeLock：handlersByType是线程不安全的容器，这个主要就是用来保证对handlersByType的操作是线程安全的，不能在post()事件的时候新增/减少事件的处理器。
• isDispatching：这个是为了防止同一个事件，多次分派给一个EventHandler执行的。

这里就不贴源码了，重点都在EventBus#register()和EventBus#post()这两个方法中了。

使用举例

 public static void main(String[] args) {

        // 同步的EventBus：post()和EventHandler的逻辑在同一个线程中执行
        EventBus eventBus = new EventBus("TestEeventBus");

        // 异步的EventBus：post()在主线程中执行，EventHandler逻辑是指定线程池驱动执行
        EventBus asyEventBus = new AsyncEventBus(new ThreadPoolExecutor(1, 1, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(128),
                new ThreadFactoryBuilder().setNameFormat("aa-%s").build(), new ThreadPoolExecutor.DiscardPolicy()));


        // 注册事件监听器
        eventBus.register(new TestEeventBusListener());

        // 注册私信监听器。
        eventBus.register(new DeadEventHandler());

        // 发布事件
        eventBus.post(new MyEvent("hello1"));
        System.out.println(Thread.currentThread().getId());

        // 这里主要就是测试验证一下：post()和时间EventHandler逻辑是在同一个线程中执行的
        new Thread(new Runnable() {
            @Override
            public void run() {
                eventBus.post(new MyEvent("hello2"));// 这个事件的EventHandler也会是在这个线程中执行
                System.out.println(Thread.currentThread().getId());
            }
        }).start();

    }
}

// 这里一个较好的实践是：Listener的名字和EventBus的idendifier保持一致。
class TestEeventBusListener {

    @Subscribe
    // 是将@Subscribe这个方法封装到了EventHandler中了，所以这个类是不需要特殊实现什么接口的。所以在register的时候直接new一个对象就好，这里new一个对象的目的是因为反射调用的时候需要有个targetObj。
    // 这里我个人觉得@Subscribe注解是使用的比较好，真的做到了解耦
    @AllowConcurrentEvents// 执行Handler逻辑的时候有synchronized锁
    public void listenMyEvent(MyEvent event) {
        System.out.println(Thread.currentThread().getName() + ":" + Thread.currentThread().getId() + "-1  :" + event.getMsg());
    }

    @Subscribe// eventBus在处理@Subscribe的时候，是以事件维度的，根这个Listener类没关系。即扫描所有的@Subscribe注解，
    // 然后以key=事件类型(@Subscribe注解的方法的入参的Class类型)，value就是这个注解的Method。
    // 所以一个Listener有多个Listener是没问题的。这种设计其实就跟TestEeventBusListener类没啥关系了。ps：我个人认为是个比较好的设计，如果让我干，我很有可能就干成了key=TestEeventBusListener了。
    public void listenMyEvent2(MyEvent2 event) {
        System.out.println(Thread.currentThread().getId() + "-2  :" + event.getMsg());
    }
}

// 私信事件监听器，在分派事件给事件监听器的时候，如果找不到事件对应的监听器，就会post()一个DeadEvent事件:DeadEvent的source封装的就是这个没有找到处理器的事件。
class DeadEventHandler {
    @Subscribe
    public void listen(DeadEvent deadEvent) {
        System.out.println(deadEvent.getSource()+""+deadEvent.getEvent());
    }
}