【Guava学习】EventBus

1，简单使用

首先引入guave的依赖；

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>23.6-jre</version>
</dependency>

再来看一下基础的使用：

EventBus eventBus = new EventBus();
eventBus.register(new Object() {
    @Subscribe
    public void hehe(String name) {
        System.out.println(name);
    }

    @Subscribe
    public void haha(Object object) {
        System.out.println(object);
    }
});
eventBus.post("huang");//输出两遍huang；
eventBus.post(17);//输出一遍17；

• 首先需要一个总线EventBus，直接new出来；
• 然后需要一个观察者，可以理解为监听器，最简洁的方式是直接new一个Object，然后向其中新增方法；
• 然后需要将观察者注册到总线中；
• 最后发布事件；

注意点：

• 使用@Subscribe注解来标注方法，该方法的返回值不会被处理，所以有无返回值都是一样的，方法的参数类型是接受的事件类型，所以方法有且只能有一个参数；
• 当多个标注了@Subscribe的方法的参数存在父子关系的时候，当发布子类型的事件时，父类型的方法也将被执行，这就是上面代码输出两遍huang的原因；
• 如果方法只标注了@Subscribe，那么该方法的执行是同步的，即使是多线程发布同一事件，那多个线程之间存在互斥锁，同一时间点，只能有一个或零个执行该方法，如下面的代码所示；

EventBus eventBus = new EventBus();
eventBus.register(new Object() {
    @Subscribe
    public void hehe(Integer num) throws InterruptedException {
        System.out.println(num + ":" + System.currentTimeMillis());
        Thread.currentThread().sleep(100);
    }
});
for (int i = 0; i < 10; i++) {
    int finalI = i;
    new Thread(new Runnable() {
        @Override
        public void run() {
            eventBus.post(finalI);
        }
    }).start();
}
//4:1517538494324
//1:1517538494425
//8:1517538494529
//0:1517538494632
//5:1517538494735
//2:1517538494838
//9:1517538494942
//6:1517538495043
//3:1517538495147
//7:1517538495250

2，关于EventBus的并行

线程安全问题是每个java程序猿都应该时刻注意的，当带有@Subscribe的方法被多个方法同时执行，且该方法内部逻辑涉及到更改成员变量时，就会出现线程安全问题，在默认情况下，如果只有@Subscribe注解时，方法时是异步执行的，即使多个线程同时调用，也需要竞争方法的同步锁，然后依次执行；

当需要@Subscribe标注的方法能被多个线程同时调用，需要配合@AllowConcurrentEvents注解使用，该注解表示允许并行执行该方法，当有多个线程同时调用方法时，因为方法无锁，所以线程可以同时进入执行，有锁和无锁，取决于@AllowConcurrentEvents，当没有该注解时，EventBus在生成Subscriber时，使用了SynchronizedSubscriber，该类型在真实调用带有@Subscribe方法时，使用了同步锁，具体后面讲解；

EventBus eventBus = new EventBus();
eventBus.register(new Object() {
    @Subscribe
    @AllowConcurrentEvents
    public void hehe(Integer num) throws InterruptedException {
        System.out.println(num + ":" + System.currentTimeMillis());
        Thread.currentThread().sleep(100);
    }
});
for (int i = 0; i < 10; i++) {
    int finalI = i;
    new Thread(new Runnable() {
        @Override
        public void run() {
            eventBus.post(finalI);
        }
    }).start();
}
//7:1517543101034
//9:1517543101034
//1:1517543101034
//4:1517543101034
//5:1517543101034
//8:1517543101034
//0:1517543101034
//6:1517543101034
//2:1517543101034
//3:1517543101034

关于多线程调用：

• 当直接new一个EventBus时，当使用post方法发布一个事件是，那么调用带有@Subscribe的方法的线程是当前线程，在哪个线程中调用，就在哪个线程中执行；
• 当需要在多线程中执行同一个标注了@Subscribe的方法时，需要使用AsyncEventBus类，并指定一个Executor线程池，那么发布事件时，调用的方法都将在指定的线程池中执行；
• 即使指定了线程池，如果没有使用@AllowConcurrentEvents，那么即使调用方法（同一方法）的线程不一样，因为同步锁的存在，执行的时机还是依次执行，多个线程并不会同时执行同一方法；
• 只有指定了线程池，并且使用了@AllowConcurrentEvents注解，才能实现在多个线程中同时调用某个标注了@Subscribe的方法，这时需要格外注意线程并发导致的线程安全问题；

EventBus eventBus = new AsyncEventBus(Executors.newFixedThreadPool(3));
eventBus.register(new Object() {
    @Subscribe
    @AllowConcurrentEvents
    public void hehe(Integer num) throws InterruptedException {
        Thread.currentThread().sleep(100);
        System.out.println(Thread.currentThread().getName() + "-" + num + "-" + System.currentTimeMillis());
    }
});
for (int i = 0; i < 9; i++) {
    eventBus.post(i);
}
//pool-1-thread-2-1-1517546343599
//pool-1-thread-1-0-1517546343599
//pool-1-thread-3-2-1517546343599
//pool-1-thread-2-3-1517546343704
//pool-1-thread-3-5-1517546343704
//pool-1-thread-1-4-1517546343704
//pool-1-thread-3-7-1517546343807
//pool-1-thread-1-8-1517546343807
//pool-1-thread-2-6-1517546343807

3，源码解析

基本组件：

• Executor：执行的线程池，默认是DirectExecutor，他未开启新的线程，而是在当前线程中直接执行；
• SubscriberRegistry：Subscriber注册器，每个带有@Subscribe的方法会被注册到该类中；
• Dispatcher：调度器，负责将事件，分发给事件对应的Subscriber，并使用Executor执行这些Subscriber；
• SubscriberExceptionHandler：异常处理器，用来处理异常；

3.1，Executor

private enum DirectExecutor implements Executor {
    INSTANCE;
    
    @Override
    public void execute(Runnable command) {
      command.run();
    }
    
    @Override
    public String toString() {
      return "MoreExecutors.directExecutor()";
    }
}

• 默认的执行器，直接实现了Executor接口，然后覆写方法，因为没有开启新的线程，所以默认是在当前线程中执行；
• 如果是在主线程中调用，那么DirectExecutor对应的线程就是主线程，如果在其他线程中执行，那么DirectExecutor对应的就是其他线程，总之，就是调用线程；

3.2，Dispatcher

private static final class PerThreadQueuedDispatcher extends Dispatcher {
    private final ThreadLocal<Queue<Event>> queue =
            new ThreadLocal<Queue<Event>>() {
                @Override
                protected Queue<Event> initialValue() {
                    return Queues.newArrayDeque();
                }
            };

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

• 该调度器名为单线程调度器，当某个线程调用dispatch方法时，首先从当前线程中获取queue，这是一个ThreadLocal类型的Event队列，然后将当前事件放置进去；
• 然后根据同样是ThreadLocal类型的dispatching字段，判断是否正在调用中，如果不是，则开始执行具体的调度任务；
  • 但是此处的if判断我不是很理解，因为结果必然一定成立；
  • 如果是单个线程，if条件中的逻辑没有执行完时，是不可能再次调用dispatch方法的；
  • 如果是多线程调用，每个线程拿到的dispatching都是不同的，相互之间不存在干扰，所以这个if条件是必然成立的，望大神排异解惑；

下面的代码是具体的调用逻辑，使用的是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));
                    }
                }
            });
}

@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.3，SubscriberRegistry

• SubscriberRegistry只有一个类就是SubscriberRegistry；
• 该类可以注册也可以取消注册Subscriber；
• 还需要注意的是，当没有@AllowConcurrentEvents注解时，Subscriber使用的是SynchronizedSubscriber类型，而有@AllowConcurrentEvents注解时，使用的是Subscriber类型；
  • 当方法被调用时都是调用invokeSubscriberMethod方法；
  • SynchronizedSubscriber类继承了Subscriber类，并重写了invokeSubscriberMethod方法；
  • 不同的是SynchronizedSubscriber类型对方法使用了同步锁，导致的结果就是，没有@AllowConcurrentEvents注解时invokeSubscriberMethod方法会在多个线程中同步执行；

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