com.google.common.eventbus.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));
}
}
});
}
用例
本文主要按照如下例子展开：
//1. 新建bus对象，默认仅能在主线程上对消息进行调度
Bus bus = new Bus(); // maybe singleton //2. 新建类A(subscriber)，answerAvailable()方法为事件回调，在主线程上运行
class A { public A() {
bus.register(this);
} // 可见性为public，仅有一个Event类型的参数
@Subscribe public void answerAvailable(AnswerAvailableEvent event) {
// process event
}
} //3. 往bus投递事件
bus.post(new AnswerAvailableEvent(42)); //4. 如果要A在注册时马上接收到一次回调，则可以新建类B(Producer)，produceAnswer()
// 方法会在注册subscriber时，对每个订阅了AnswerAvailableEvent方法发送事件
class B { public B() {
bus.register(this);
} //可见性为public，不带任何参数
@Produce public AnswerAvailableEvent produceAnswer() {
return new AnswerAvailableEvent();
}
}
初始化
首先来看看Bus bus = new Bus()这一句，对应的源码如下所示：
public Bus() {
this(DEFAULT_IDENTIFIER);
}
public Bus(String identifier) {
this(ThreadEnforcer.MAIN, identifier);
}
public Bus(ThreadEnforcer enforcer, String identifier) {
this(enforcer, identifier, HandlerFinder.ANNOTATED);
}
Bus(ThreadEnforcer enforcer, String identifier, HandlerFinder handlerFinder) {
this.enforcer = enforcer;
this.identifier = identifier;
this.handlerFinder = handlerFinder;
}
默认参数为enforcer = ThreadEnforcer.MAIN，identifier = DEFAULT_IDENTIFIER，handlerFinder = HandlerFinder.ANNOTATED。我们来看看这些参数是什么意思。
ThreadEnforcer
ThreadEnforcer是一个接口，enforce()方法用于检查当前线程是否为指定的线程类型：
public interface ThreadEnforcer {
ThreadEnforcer ANY = new ThreadEnforcer() {
@Override
public void enforce(Bus bus) {
// Allow any thread.
}
};
ThreadEnforcer MAIN = new ThreadEnforcer() {
@Override
public void enforce(Bus bus) {
if (Looper.myLooper() != Looper.getMainLooper()) {
throw new IllegalStateException("Event bus " + bus +
" accessed from non-main thread " + Looper.myLooper());
}
}
};
void enforce(Bus bus);
}
不带参数的构造函数bus()使用默认的ThreadEnforcer.MAIN，表示enforce()方法必须在主线程上执行。
identifier
identifier仅为bus的名字，debug用。
handlerFinder
HandlerFinder用于在注册/反注册的时候查找Subscriber和Producer，后文会对其展开源码级别的解析。不带参数的构造函数bus()使用默认的HandlerFinder.ANNOTATED，表示使用注解来进行查找。
除上述以外，bus类还有两个成员变量handlersByType和producersByType：
private final ConcurrentMap<Class<?>, Set<EventHandler>> handlersByType =
new ConcurrentHashMap<Class<?>, Set<EventHandler>>();
private final ConcurrentMap<Class<?>, EventProducer> producersByType =
new ConcurrentHashMap<Class<?>, EventProducer>();
分别用于通过event的类型（class类型）来查找event handle和event producer。
注册/反注册事件
如下所示，要A成为订阅者订阅AnswerAvailableEvent，只需将其注册到bus，然后使用@Subscribe注解标记回调方法即可。回调方法要求可见性为public，有且仅有一个参数，类型为订阅的event。
class A {
public A() {
bus.register(this);
}
@Subscribe public void answerAvailable(AnswerAvailableEvent event) {
// process event
}
}
@Subscribe
首先看一下@Subscribe注解：
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface Subscribe {
}
RetentionPolicy.RUNTIME表示它是运行时的注解，ElementType.METHOD表示用于注解方法。
bus.register
再看一下register流程：
public void register(Object object) {
if (object == null) {
throw new NullPointerException("Object to register must not be null.");
}
//1. 检查当前线程是否符合ThreadEnforcer的设置
enforcer.enforce(this);
//2. 默认情况下，通过注解在object上找出所有Producer
Map<Class<?>, EventProducer> foundProducers = handlerFinder.findAllProducers(object);
for (Class<?> type : foundProducers.keySet()) {
//2-1. 查一下object上的producer注册的event是否已经被别人注册过。
final EventProducer producer = foundProducers.get(type);
EventProducer previousProducer = producersByType.putIfAbsent(type, producer);
//checking if the previous producer existed
if (previousProducer != null) {
throw new IllegalArgumentException("Producer method for type " + type + " found on type " + producer.target.getClass() + ", but already registered by type " + previousProducer.target.getClass() + ".");
}
//2-2. 如果没有注册过，那么找出对应event的handler，触发一次回调。
Set<EventHandler> handlers = handlersByType.get(type);
if (handlers != null && !handlers.isEmpty()) {
for (EventHandler handler : handlers) {
dispatchProducerResultToHandler(handler, producer);
}
}
}
//3. 找出object上用@Subscribe注解了的方法
Map<Class<?>, Set<EventHandler>> foundHandlersMap = handlerFinder.findAllSubscribers(object);
for (Class<?> type : foundHandlersMap.keySet()) {
Set<EventHandler> handlers = handlersByType.get(type);
if (handlers == null) {
//3-1. 该event是第一次注册，那么新建一个CopyOnWriteArraySet用来保存handler和event的对应关系（EventHandler）
//concurrent put if absent
Set<EventHandler> handlersCreation = new CopyOnWriteArraySet<EventHandler>();
handlers = handlersByType.putIfAbsent(type, handlersCreation);
if (handlers == null) {
handlers = handlersCreation;
}
}
//3-2. 保存object中新增的event-handler对应关系。
final Set<EventHandler> foundHandlers = foundHandlersMap.get(type);
if (!handlers.addAll(foundHandlers)) {
throw new IllegalArgumentException("Object already registered.");
}
}
//4. 检查object上的event是否存在对应的Producer，有则触发一次回调
for (Map.Entry<Class<?>, Set<EventHandler>> entry : foundHandlersMap.entrySet()) {
Class<?> type = entry.getKey();
EventProducer producer = producersByType.get(type);
if (producer != null && producer.isValid()) {
Set<EventHandler> foundHandlers = entry.getValue();
for (EventHandler foundHandler : foundHandlers) {
if (!producer.isValid()) {
break;
}
if (foundHandler.isValid()) {
dispatchProducerResultToHandler(foundHandler, producer);
}
}
}
}
}
总的来说register做了三件事情：触发新的Producer；注册新的event-handler关系；触发旧的Producer。另外有两点要注意一下：
由于在一般使用场景下，发送/处理event远比注册/反注册操作频繁，所以在保证线程安全的情况下，使用CopyOnWriteArraySet作为保存event和handler的容器，可以大大提高效率。
CopyOnWrite容器在读的时候不会加锁，写的时候先复制一份，写完再替换原容器。如果容器正在写操作时发生了读操作（或者正在读的时候发生了写操作），读操作的对象为容器的快照（snapshot）。由于register方法没有加锁，所以在3-1中，尽管已经检查了handlers是否存在，但仍需使用putIfAbsent来保存handler。
EventProducer和EventHandler
注意到bus通过HandlerFinder来查找object上的producer和subscriber，接下来看一下HandlerFinder的实现：
interface HandlerFinder {
HandlerFinder ANNOTATED = new HandlerFinder() {
@Override
public Map<Class<?>, EventProducer> findAllProducers(
Object listener) {
return AnnotatedHandlerFinder.findAllProducers(listener);
}
@Override
public Map<Class<?>, Set<EventHandler>> findAllSubscribers(
Object listener) {
return AnnotatedHandlerFinder.findAllSubscribers(listener);
}
};
Map<Class<?>, EventProducer> findAllProducers(Object listener);
Map<Class<?>, Set<EventHandler>> findAllSubscribers(Object listener);
}
其中findAllProducers方法返回某event type对应的EventProducer，findAllSubscribers返回某event type对应的EventHandler集合。先看一下EventProducer和EventHandler。
EventProducer是一个producer方法的包装类，源码如下：
class EventProducer {
final Object target;
private final Method method;
private final int hashCode;
private boolean valid = true;
EventProducer(Object target, Method method) {
if (target == null) {
throw new NullPointerException(
"EventProducer target cannot be null.");
}
if (method == null) {
throw new NullPointerException(
"EventProducer method cannot be null.");
}
this.target = target;
this.method = method;
method.setAccessible(true);
// 提前计算hashcode，以防每次调用hash()时消耗资源
final int prime = 31;
hashCode = ((prime + method.hashCode()) * prime) + target.hashCode();
}
public boolean isValid() {
return valid;
}
// 应在object unregister时调用
public void invalidate() {
valid = false;
}
public Object produceEvent() throws InvocationTargetException {
if (!valid) {
throw new IllegalStateException(toString() +
" has been invalidated and can no longer produce events.");
}
try {
return method.invoke(target);
} catch (IllegalAccessException e) {
throw new AssertionError(e);
} catch (InvocationTargetException e) {
if (e.getCause() instanceof Error) {
throw (Error) e.getCause();
}
throw e;
}
}
}
其中produceEvent方法用于获得event。可以看出为什么Otto要求produce函数不能有参数。
与EventProducer类似，EventHandler是一个event handler方法（事件回调）的包装类，源码如下：
class EventHandler {
private final Object target;
private final Method method;
private final int hashCode;
private boolean valid = true;
EventHandler(Object target, Method method) {
if (target == null) {
throw new NullPointerException(
"EventHandler target cannot be null.");
}
if (method == null) {
throw new NullPointerException(
"EventHandler method cannot be null.");
}
this.target = target;
this.method = method;
method.setAccessible(true);
// Compute hash code eagerly since we know it will be used frequently and we cannot estimate the runtime of the
// target's hashCode call.
final int prime = 31;
hashCode = ((prime + method.hashCode()) * prime) + target.hashCode();
}
public boolean isValid() {
return valid;
}
public void invalidate() {
valid = false;
}
public void handleEvent(Object event) throws InvocationTargetException {
if (!valid) {
throw new IllegalStateException(toString() +
" has been invalidated and can no longer handle events.");
}
try {
method.invoke(target, event);
} catch (IllegalAccessException e) {
throw new AssertionError(e);
} catch (InvocationTargetException e) {
if (e.getCause() instanceof Error) {
throw (Error) e.getCause();
}
throw e;
}
}
}
其中handleEvent方法用于在object上调用handle方法（事件回调），传入event对象。可以看出为什么Otto要求event handler函数仅能有一个参数。
dispatchProducerResultToHandler
dispatchProducerResultToHandler方法用于将Producer产生的event分发给对应的handler。源码如下所示：
private void dispatchProducerResultToHandler(EventHandler handler, EventProducer producer) {
Object event = null;
try {
event = producer.produceEvent();
} catch(InvocationTargetException e) {
throwRuntimeException("Producer " + producer + " threw an exception.", e);
}
if (event == null) {
return;
}
dispatch(event, handler);
}
protected void dispatch(Object event, EventHandler wrapper) {
try {
wrapper.handleEvent(event);
} catch(InvocationTargetException e) {
throwRuntimeException("Could not dispatch event: " + event.getClass() + " to handler " + wrapper, e);
}
}
逻辑比较简单，主要是使用了Producer的produceEvent()方法获得event对象后，调用EventHandler的handleEvent()方法。
bus.unregister
Bus类的unregister方法用于解除目标对象和bus之间的关联关系，包括对象上的producer方法，subscriber方法，源码如下所示：
public void unregister(Object object) {
if (object == null) {
throw new NullPointerException("Object to unregister must not be null.");
}
//1. 检查当前线程是否符合ThreadEnforcer的设置
enforcer.enforce(this);
//2. 默认情况下，通过注解在object上找出所有Producer，将其从producersByType中删除并标记为invalidate
Map<Class<?>, EventProducer> producersInListener = handlerFinder.findAllProducers(object);
for (Map.Entry<Class<?>, EventProducer> entry : producersInListener.entrySet()) {
final Class<?> key = entry.getKey();
EventProducer producer = getProducerForEventType(key);
EventProducer value = entry.getValue();
if (value == null || !value.equals(producer)) {
throw new IllegalArgumentException(
"Missing event producer for an annotated method. Is " + object.getClass() + " registered?");
}
producersByType.remove(key).invalidate();
}
//3. 默认情况下，找出object上用@Subscribe注解了的handler，将其从event集合中删除并标记为invalidate
Map<Class<?>, Set<EventHandler>> handlersInListener = handlerFinder.findAllSubscribers(object);
for (Map.Entry<Class<?>, Set<EventHandler>> entry : handlersInListener.entrySet()) {
Set<EventHandler> currentHandlers = getHandlersForEventType(entry.getKey());
Collection<EventHandler> eventMethodsInListener = entry.getValue();
if (currentHandlers == null || !currentHandlers.containsAll(eventMethodsInListener)) {
throw new IllegalArgumentException(
"Missing event handler for an annotated method. Is " + object.getClass() + " registered?");
}
for (EventHandler handler : currentHandlers) {
if (eventMethodsInListener.contains(handler)) {
handler.invalidate();
}
}
currentHandlers.removeAll(eventMethodsInListener);
}
}
投递事件
一次简单的事件投递操作如下所示：
bus.post(new AnswerAvailableEvent(42));
我们来看一下post方法的源码实现：
public void post(Object event) {
if (event == null) {
throw new NullPointerException("Event to post must not be null.");
}
//1. 检查当前线程是否符合ThreadEnforcer的设置
enforcer.enforce(this);
//2. 向上追溯event的所有父类
Set<Class<?>>dispatchTypes = flattenHierarchy(event.getClass());
//3. 当前event没有注册handler，则发送一个DeadEvent事件
boolean dispatched = false;
for (Class<?>eventType: dispatchTypes) {
Set<EventHandler> wrappers = getHandlersForEventType(eventType);
if (wrappers != null && !wrappers.isEmpty()) {
dispatched = true;
for (EventHandler wrapper: wrappers) {
//3-1 将事件和handler放到分发队列里
enqueueEvent(event, wrapper);
}
}
}
//4. 当前event没有注册handler，则发送一个DeadEvent事件
if (!dispatched && !(event instanceof DeadEvent)) {
post(new DeadEvent(this, event));
}
//5. 通知队列进行分发操作
dispatchQueuedEvents();
}
注意几点：
发送一个Event时，订阅了Event父类的Subscriber方法也会被调用。
事件会被放到调用者所在线程的队列里依次分发。
下面分点进行详述。
flattenHierarchy
进行post操作时，首先会通过flattenHierarchy方法获得event的父类或者接口：
Set<Class<?>>flattenHierarchy(Class<?>concreteClass) {
Set<Class<?>>classes = flattenHierarchyCache.get(concreteClass);
if (classes == null) {
Set<Class<?>>classesCreation = getClassesFor(concreteClass);
classes = flattenHierarchyCache.putIfAbsent(concreteClass, classesCreation);
if (classes == null) {
classes = classesCreation;
}
}
return classes;
}
private Set<Class<?>> getClassesFor(Class<?> concreteClass) {
List<Class<?>> parents = new LinkedList<Class<?>>();
Set<Class<?>> classes = new HashSet<Class<?>>();
parents.add(concreteClass);
//深度优先遍历
while (!parents.isEmpty()) {
Class<?> clazz = parents.remove(0);
classes.add(clazz);
Class<?> parent = clazz.getSuperclass();
if (parent != null) {
parents.add(parent);
}
}
return classes;
}
从上可知flattenHierarchy()通过getClassesFor()利用深度优先遍历导出了concreteClass的所有父类。
Dispatch Queue
通过post方法投递的event首先会放在当前线程所在的Dispatch Queue中，然后依次分发。Bus类有如下成员属性：
private final ThreadLocal<ConcurrentLinkedQueue<EventWithHandler>> eventsToDispatch =
new ThreadLocal<ConcurrentLinkedQueue<EventWithHandler>>() {
@Override protected ConcurrentLinkedQueue<EventWithHandler> initialValue() {
return new ConcurrentLinkedQueue<EventWithHandler>();
}
};
eventsToDispatch是一个ThreadLocal对象，通过initialValue()方法，eventsToDispatch每次在新的线程上调用的时候都会生成新的ConcurrentLinkedQueue实例。event是通过enqueueEvent(event, wrapper)方法放到queue中的，下面看看enqueueEvent()的实现：
protected void enqueueEvent(Object event, EventHandler handler) {
eventsToDispatch.get().offer(new EventWithHandler(event, handler));
}
offer()方法会会将EventWithHandler对象放到当前线程的queue的尾部。offer方法和add方法的区别在于，当无法插入（例如空间不够）的情况发生时会发挥false，热不是抛出异常。EventWithHandler类对event和handler的关系进行了简单的包装，实现如下：
static class EventWithHandler {
final Object event;
final EventHandler handler;
public EventWithHandler(Object event, EventHandler handler) {
this.event = event;
this.handler = handler;
}
}
接下来看看dispatchQueuedEvents方法的实现：
protected void dispatchQueuedEvents() {
// don't dispatch if we're already dispatching, that would allow reentrancy and out-of-order events. Instead, leave
// the events to be dispatched after the in-progress dispatch is complete.
//1. 不能重复分发，否则会导致event的分发次序混乱
if (isDispatching.get()) {
return;
}
isDispatching.set(true);
try {
while (true) {
//2. 依次取出EventWithHandler，并通过dispatch方法进行分发。
EventWithHandler eventWithHandler = eventsToDispatch.get().poll();
if (eventWithHandler == null) {
break;
}
if (eventWithHandler.handler.isValid()) {
dispatch(eventWithHandler.event, eventWithHandler.handler);
}
}
} finally {
isDispatching.set(false);
}
}
值得注意的是，所有subscribe方法抛出的异常都会在这里捕获，捕获到异常以后event分发过程即停止，直到下一次在该线程上调用post为止。
结构图
综上，Otto的总体结构可用下图表示：
+-------------------------+
|Bus(ThreadLocal) |
| +--------------+ |
| |EventProducers| |
| | +-------+ | register +-------+
| | |Produce| <----+-------+Produce|
| | +-------+ | | +-------+
| | +-------+ | |
| | |Produce| | |
| | +-------+ | |
| +--------------+ |
| | |
| event |
| | |
post(event)| +-------v--------+ |
+----------------> Dispatch Queue | |
| +-------+--------+ |
| | |
| event |
| | |
| +------v------+ |
| |EventHandlers| |
| | +---------+ | |
| | |Subscribe| | register +---------+
| | +---------+ <-----+-------+Subscribe|
| | +---------+ | | +---------+
| | |Subscribe| | |
| | +---------+ | |
| +-------------+ |
| |
+-------------------------+https://segmentfault.com/a/1190000003982257