基于版本：Guava 22.0
Wiki：EventBus
0. EventBus简介
提供了发布-订阅模型，可以方便的在EventBus上注册订阅者，发布者可以简单的将事件传递给EventBus，EventBus会自动将事件传递给相关联的订阅者。
支持同步/异步模式。
只能用于线程间通信。
1. EventBus类图
EventBus是同步实现
AsyncEventBus是异步实现
2. 代码实例
EventBus eventBus = new EventBus();
eventBus.register(new Object() {
@Subscribe
public void listen(Object subReport) throws Exception {
System.out.println("receive object event!");
}
@Subscribe
public void listen(Integer subReport) throws Exception {
System.out.println("receive integer event!");
}
});
eventBus.post(Integer.valueOf(1));
这段代码的输出如下：
receive integer event!
receive object event!
可以看到我们声明了一个EventBus，然后向其注册了两个订阅者（含有Subscribe注解的方法），然后调用post方法向EventBus发布了一条消息。
消息类型是Integer，由于订阅者的关注对象是Integer与Object，都与这条消息有关，所以两个订阅者都收到了通知。
但是这个发布-订阅模式是如何实现的呢？我们下面来逐步分析一下EventBus的源码。
3. EventBus.register()
EventBus.register
/**
* Registers all subscriber methods on {@code object} to receive events.
*
* @param object object whose subscriber methods should be registered.
*/
public void register(Object object) {
subscribers.register(object);
}
SubscriberRegistry.register
/**
* Registers all subscriber methods on the given listener object.
*/
void register(Object listener) {
Multimap<Class<?>, Subscriber> listenerMethods = findAllSubscribers(listener);//获取传入的listener中含有的所有的订阅者
for (Map.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<Subscriber>();
eventSubscribers =
MoreObjects.firstNonNull(subscribers.putIfAbsent(eventType, newSet), newSet);//线程安全的修改subscribers对象
}
eventSubscribers.addAll(eventMethodsInListener);
}
}
/**
* Returns all subscribers for the given listener grouped by the type of event they subscribe to.
分析传入的对象，遍历其中所有含有Subscribe注解而且只含有一个参数的方法，然后将其包装成订阅者并返回。
*/
private Multimap<Class<?>, Subscriber> findAllSubscribers(Object listener) {
Multimap<Class<?>, Subscriber> methodsInListener = HashMultimap.create();
Class<?> clazz = listener.getClass();
for (Method method : getAnnotatedMethods(clazz)) {//getAnnotatedMethods方法最终会调用到下面的SubscriberRegistry.getAnnotatedMethodsNotCached方法，这个方法会用反射处理传入的clazz及其所有的父类，提取出含有Subscribe注解并且有且只有一个参数的方法
Class<?>[] parameterTypes = method.getParameterTypes();
Class<?> eventType = parameterTypes[0];//获取这个方法的唯一参数的Class
methodsInListener.put(eventType, Subscriber.create(bus, listener, method));//将EventBus，订阅者对象，订阅者方法包装一下并放入map，后续触发事件的时候会用到
}
return methodsInListener;
}
SubscriberRegistry.getAnnotatedMethodsNotCached
private static ImmutableList<Method> getAnnotatedMethodsNotCached(Class<?> clazz) {
Set<? extends Class<?>> supertypes = TypeToken.of(clazz).getTypes().rawTypes();//连clazz的父类也会处理
Map<MethodIdentifier, Method> identifiers = Maps.newHashMap();
for (Class<?> supertype : supertypes) {
for (Method method : supertype.getDeclaredMethods()) {
if (method.isAnnotationPresent(Subscribe.class) && !method.isSynthetic()) {//含有Subscribe注解，而且不是合成的方法
// 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);//方法必须有且只有一个参数
MethodIdentifier ident = new MethodIdentifier(method);
if (!identifiers.containsKey(ident)) {
identifiers.put(ident, method);
}
}
}
}
return ImmutableList.copyOf(identifiers.values());
}
从SubscriberRegistry.getAnnotatedMethods到SubscriberRegistry.getAnnotatedMethodsNotCached的跳转会涉及到Guava的LoadingCache，其他的逻辑都非常直观。
register方法的大概流程如下：
a. 解析传入的Object对象，用反射分析对应的class及其所有父类，找到所有用Subscribe注解修饰，而且只有一个参数的方法。由这个方法可以生成订阅者对象。
b. 解析这个方法的参数（订阅者订阅的事件类型）
c. 每个EventBus在初始化时都会与一个SubscriberRegistry关联，这个SubscriberRegistry内部维护了一个名为subscribers的ConcurrentMap，这个ConcurrentMap的 key是EventBus关心的事件类型，value是订阅了这些事件的订阅者的集合，subscribers在后续发布事件的流程中非常有用（可以通过发布的事件类型，找到这个事件所关联的所有订阅者，并通知这些相关的订阅者）。现在用a,b中的信息来更新subscribers，由于subscribers可能被并发操作，所以用到了ConcurrentMap.putIfAbsent方法以保证线程安全。
总之，调用register方法会更新subscribers，subscribers中含有事件与订阅者的关联关系。
4. EventBus.post()
EventBus.post
public void post(Object event) {
Iterator<Subscriber> eventSubscribers = subscribers.getSubscribers(event);//获取发布事件所关联的所有订阅者（包括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));//如果这个事件没有订阅者，则将这个事件包装为DeadEvent事件，然后重新触发post方法
}
}
SubscriberRegistry.getSubscribers
/**
* Gets an iterator representing an immutable snapshot of all subscribers to the given event at
* the time this method is called.
*/
Iterator<Subscriber> getSubscribers(Object event) {
ImmutableSet<Class<?>> eventTypes = flattenHierarchy(event.getClass());//flattenHierarchy方法会解析出event的class，及其所有父类/接口的class
List<Iterator<Subscriber>> subscriberIterators =
Lists.newArrayListWithCapacity(eventTypes.size());
for (Class<?> eventType : eventTypes) {//遍历event关联的所有class，找到这些class关联的所有订阅者，添加到subscriberIterators中
CopyOnWriteArraySet<Subscriber> eventSubscribers = subscribers.get(eventType);
if (eventSubscribers != null) {
// eager no-copy snapshot
subscriberIterators.add(eventSubscribers.iterator());
}
}
return Iterators.concat(subscriberIterators.iterator());
}
SubscriberRegistry.flattenHierarchy
/**
* Flattens a class's type hierarchy into a set of {@code Class} objects including all
* superclasses (transitively) and all interfaces implemented by these superclasses.
*/
@VisibleForTesting
static ImmutableSet<Class<?>> flattenHierarchy(Class<?> concreteClass) {
try {
return flattenHierarchyCache.getUnchecked(concreteClass);//这里会用Guava的TypeToken工具解析出concreteClass的class，以及所有父类/接口的class并返回
} catch (UncheckedExecutionException e) {
throw Throwables.propagate(e.getCause());
}
}
post方法的大概流程如下：
a. 用Guava的TypeToken工具解析Event的class，以及Event的所有父类/接口的class，所有订阅了这些class的订阅者后续都会收到通知
b. 调用EventBus关联的Dispatcher的dispatcher方法，将Event发布给所有关联的订阅者。
下一步我们会解析Dispatcher，以及Dispatcher是如何实现EventBus的同步/异步语义的。
5. EventBus的默认Dispatcher
EventBus
/**
* Creates a new EventBus named "default".
*/
public EventBus() {
this("default");
}
/**
* 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,
MoreExecutors.directExecutor(),//由当前线程直接运行提交任务的“线程池”
Dispatcher.perThreadDispatchQueue(),
LoggingHandler.INSTANCE);
}
Dispatcher.perThreadDispatchQueue()
static Dispatcher perThreadDispatchQueue() {
return new PerThreadQueuedDispatcher();
}
Dispatcher.PerThreadQueuedDispatcher
/**
* Implementation of a {@link #perThreadDispatchQueue()} dispatcher.
*/
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();
}
};//为每个调用post方法的工作线程维护一个发布队列，工作线程独立操作这个队列完成时间发布流程，所以是线程安全的
/**
* 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));//将发布事件+事件所关联的所有订阅者包装成一个Event对象，并提交至发布队列
if (!dispatching.get()) {
dispatching.set(true);
try {
Event nextEvent;
while ((nextEvent = queueForThread.poll()) != null) {//尝试从发布队列中poll元素
while (nextEvent.subscribers.hasNext()) {//遍历发布事件的所有订阅者，向订阅者派发事件，由于EventBus默认使用的线程池是MoreExecutors.directExecutor()，所以实际上发布者会串行而且同步的向事件的所有订阅者派发事件，直到全部派发结束，post方法才会返回，所以EventBus在默认情况下是同步的。
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;
}
}
}
Subscriber.dispatchEvent
/**
* 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));
}
}
});
}
Subscriber.invokeSubscriberMethod
/**
* Invokes the subscriber method. This method can be overridden to make the invocation
* synchronized.
*/
@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;
}
}
同步语义的实现关键在于
a. 使用的线程池为MoreExecutors.directExecutor()，这实际上不能算是一个真正的线程池，所有提交到这个池子里的任务，都是由提交任务的线程自身来执行的。
b. EventBus为每个调用post方法的线程维护了一个发布队列，工作线程会将事件提交到这个私有队列里，然后逐个通知事件所关联的订阅者，由于使用的线程池是MoreExecutors.directExecutor()，所以这个过程实际上是完全串行而且同步执行的，调用post方法的工作线程实际上会在将事件通知给所有订阅者后才会返回，从而实现了同步的EventBus语义。
6. AsyncEventBus的Dispatcher
AsyncEventBus
public AsyncEventBus(Executor executor) {
super("default", executor, Dispatcher.legacyAsync(), LoggingHandler.INSTANCE);
Dispatcher.legacyAsync()
static Dispatcher legacyAsync() {
return new LegacyAsyncDispatcher();
}
Dispatcher.LegacyAsyncDispatcher
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()) {//将Event+订阅者包装成EventWithSubscriber，放入发布队列中
queue.add(new EventWithSubscriber(event, subscribers.next()));
}
EventWithSubscriber e;
while ((e = queue.poll()) != null) {
e.subscriber.dispatchEvent(e.event);//从发布队列中提取EventWithSubscriber，然后将事件发布给订阅者，这一过程是发送到AsyncEventBus初始化时提供的线程池里执行的，所以是异步的。也就是说post方法返回的时候，可能发布过程尚未完成，还有订阅者没有收到消息。
}
}
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;
}
}
}
异步语义实现的关键在于：
a. 发布事件使用的线程池是创建AsyncEventBus时传入的线程池，一般来说都是正常的含有多个工作线程的线程池，提交到线程池里的任务可以被异步执行
b. LegacyAsyncDispatcher中维护了一个公用的ConcurrentLinkedQueue，这个AsyncEventBus收到的所有Event都会被put到这个队列中，put完Event的线程同时也会从这个队列中poll Event，然后submit这些Event到线程池中，如果这个线程池是普通的含有工作线程的线程池，那么submit完Event之后post方法即可立即返回，传递Event给订阅者的任务会由线程池里的工作线程完成。这样就实现了异步语义。
总的来说，EventBus的源码还是比较清晰易懂的，实现手法也非常优雅，值得我们学习。