一、问题描述
在工作中，经常会遇见使用异步的方式来发送事件，或者触发另外一个动作：经常用到的框架是MQ（分布式方式通知）。如果是同一个jvm里面通知的话，就可以使用EventBus。由于EventBus使用起来简单、便捷，因此，工作中会经常用到。深入理解该框架的原理就很有必要。
二、框架解析
2.1、组织结构
eventbus的组织结构如下：
eventbus主要有以下几部分组成：
1、eventbus、asyncEventBus：事件发送器。
2、event：事件承载单元。
3、SubscriberRegistry：订阅者注册器，将订阅者注册到event上，即将有注解Subscribe的方法和event绑定起来。
4、Dispatcher：事件分发器，将事件的订阅者调用来执行。
5、Subscriber、SynchronizedSubscriber：订阅者，并发订阅还是同步订阅。
2.2、运行原理
1、eventbus是基于注册监听的方式来运行的，因此，首先需要将eventbus，然后才会有事件及监听者。新建eventbus或者AsyncEventBus的方式如下：
EventBus eventBus = new EventBus();或者
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(20);
ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 20,
30, TimeUnit.SECONDS, workQueue);
AsyncEventBus asyncEventBus = new AsyncEventBus(executor);2、注册监听者。
eventBus.register(eventListener);底层就是将类eventListener中所有注解有Subscribe的方法与其Event对放在一个map中（一个event可以对应多个Subscribe的方法）。实现如下：
void register(Object listener) {
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<>();
eventSubscribers =
MoreObjects.firstNonNull(subscribers.putIfAbsent(eventType, newSet), newSet);
}
eventSubscribers.addAll(eventMethodsInListener);
}
}3、事件发送：执行指定事件类型的订阅者（包含了method），从订阅者中获取指定事件的订阅者，然后按照规则（同步、异步）执行指定的方法。
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));
}
}上述代码说明，如果事件没有监听者，就当作死亡事件来对待。
/** 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));
}
}
});
}
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;
}
}这里就说明，最后就是被订阅的方法被调用。
4、EventBus与AsyncEventBus的区别
从字面上看，AsyncEventBus是异步的EventBus，那么EventBus应该就是同步的了。EventBus的executor为MoreExecutors.directExecutor()，其实现如下：
public static Executor directExecutor() {
return DirectExecutor.INSTANCE;
}
/** See {@link #directExecutor} for behavioral notes. */
private enum DirectExecutor implements Executor {
INSTANCE;
@Override
public void execute(Runnable command) {
command.run();
}
@Override
public String toString() {
return "MoreExecutors.directExecutor()";
}
}其execute方法直接执行线程的run方法，即同步调用run方法执行。EventBus的dispatcher为PerThreadQueuedDispatcher。其dispatch方法如下：
@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();
}
}
}dispatchEvent的实现如下：
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));
}
}
});
}因此，整个执行过程如下：
整个过程都是同步方式执行，因此，EventBus是同步的。
AsyncEventBus的dispatcher为LegacyAsyncDispatcher，executor为自己指定的线程池。运行流程如下：
虚线为线程池异步调度，因此，AsyncEventBus为异步方式。
5、AllowConcurrentEvents的作用
它所在的代码为：
static Subscriber create(EventBus bus, Object listener, Method method) {
return isDeclaredThreadSafe(method)
? new Subscriber(bus, listener, method)
: new SynchronizedSubscriber(bus, listener, method);
}
private static boolean isDeclaredThreadSafe(Method method) {
return method.getAnnotation(AllowConcurrentEvents.class) != null;
}即如果订阅者方法上有注解AllowConcurrentEvents，则返回Subscriber，否则，返回SynchronizedSubscriber。SynchronizedSubscriber的字面意思为同步订阅者，它的实现代码为：
@Override
void invokeSubscriberMethod(Object event) throws InvocationTargetException {
synchronized (this) {
super.invokeSubscriberMethod(event);
}
}即没有使用注解AllowConcurrentEvents的订阅者，在并发环境中，都是串行执行。这在高并发环境中，会严重影响性能。
三、使用案例
3.1、eventbus定义
@Configuration
public class ConfigBean {
@Bean
public EventBus executorService() {
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(20);
ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 20,
30, TimeUnit.SECONDS, workQueue);
return new AsyncEventBus(executor);
}
}3.2、注册与事件发送
@Service
public class TestService implements InitializingBean {
@Autowired
private EventListener eventListener ;
@Autowired
private EventBus eventBus ;
public void postEvent(){
eventBus.post(new LoginEvent("iwill","123456"));
}
@Override
public void afterPropertiesSet() throws Exception {
eventBus.register(eventListener);
}
}3.3、订阅者定义
package com.iwill.eventBus.listener;
import com.google.common.eventbus.Subscribe;
import com.iwill.eventBus.event.LoginEvent;
import com.iwill.eventBus.event.RegisterEvent;
import org.springframework.stereotype.Component;
@Component
public class EventListener {
@Subscribe
public void subscribeLoginEvent1(LoginEvent event){
System.out.println("method 1 : receive login event ");
}
@Subscribe
public void subscribeLoginEvent2(LoginEvent event){
System.out.println("method 2 : receive login event ");
}
@Subscribe
public void subscribeRegisterEvent(RegisterEvent event){
try{
Thread.sleep(10000L);
}catch (Exception exp){
exp.printStackTrace();
}
System.out.println("method : receive register event ");
}
}四、注意事项
1、在高并发的环境下使用AsyncEventBus时，发送事件可能会出现异常，因为它使用的线程池，当线程池的线程不够用时，会拒绝接收任务，就会执行线程池的拒绝策略，如果需要关注是否提交事件成功，就需要将线程池的拒绝策略设为抛出异常，并且try-catch来捕获异常。如下：
try {
eventBus.post(new LoginEvent("iwill", "123456"));
}catch (Exception exp){
//TODO 落表或者其他处理
}2、本文用到的guava版本如下：
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>26.0-jre</version>
</dependency>本文来自云栖社区合作伙伴“开源中国”
本文作者：王练
