本文主要按照以下例子展开:安全
//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()
这一句,对应的源码以下所示:app
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。咱们来看看这些参数是什么意思。ide
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()方法必须在主线程上执行。oop
identifier仅为bus的名字,debug用。post
HandlerFinder用于在注册/反注册的时候查找Subscriber和Producer,后文会对其展开源码级别的解析。不带参数的构造函数bus()使用默认的HandlerFinder.ANNOTATED,表示使用注解来进行查找。this
除上述之外,bus类还有两个成员变量handlersByType和producersByType:spa
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。debug
class A { public A() { bus.register(this); } @Subscribe public void answerAvailable(AnswerAvailableEvent event) { // process event } }
首先看一下@Subscribe注解:
@Retention(RetentionPolicy.RUNTIME) @Target(ElementType.METHOD) public @interface Subscribe { }
RetentionPolicy.RUNTIME表示它是运行时的注解,ElementType.METHOD表示用于注解方法。
再看一下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。
注意到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方法用于将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之间的关联关系,包括对象上的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方法也会被调用。
事件会被放到调用者所在线程的队列里依次分发。
下面分点进行详述。
进行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的全部父类。
经过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| | | | | +---------+ | | | +-------------+ | | | +-------------------------+