Otto源碼分析

Otto是一個輕量級的EventBus，它的使用非常簡單，我們使用一個Bus的單例，所有需要產生事件(@Produce bus.post(new YourEvent(…)))或者處理事件(@Subscribe)的對象，在create時register，銷毀destroy時unregister即可。

使用

@Subscribe
訂閱事件，也就是事件的處理者，它有且僅有一個參數YourEvent，每一個Subscribe對應處理一個YourEvent。Event用於連接（匹配）post和訂閱。@Subscribe使用舉例：

@Subscribe
public void reveiverMethod(YourEvent event){
   //...TODO 
}

@Produce
產生事件，改方法在對象被register後即被調用（–使用情況比較特殊的），該方法必須有一個非空的返回值，參數必須為空。 bus.post(new YourEvent(…))
發送一個事件，等待@Subcribe處理

使用舉例

MainActivity

package com.example.net.mobctrl.ottotest;

import com.squareup.otto.Produce;
import com.squareup.otto.Subscribe;

import android.app.Activity;
import android.os.Bundle;
import android.view.View;
import android.view.View.OnClickListener;
import android.widget.TextView;

public class MainActivity extends Activity {

    TextView tvShow;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        BusManager.getInstance().register(this);
        System.out.println(debug:onCreate);
        setContentView(R.layout.activity_main);
        findViewById(R.id.btn_1).setOnClickListener(new OnClickListener() {

            @Override
            public void onClick(View arg0) {
                BusManager.getInstance().post(new MyEvent(將我點擊的內容，發送出去));
            }
        });
        tvShow = (TextView) findViewById(R.id.tv_show);
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        BusManager.getInstance().unregister(this);
    }

    @Subscribe
    public void receiveEventByParam(MyEvent event) {
        System.out.println(debug: + event.getContent());
        if (tvShow != null) {
            tvShow.setText(event.getContent());
        }
    }

    @Produce
    public MyEvent sendEvent() {
        return new MyEvent(這是我產生的事件(@Produce));
    }
}

BusManager 是一個單例

package com.example.net.mobctrl.ottotest;

import com.squareup.otto.Bus;
public class BusManager {
    private static Bus bus = null;
    private BusManager() {
    }
    public static synchronized Bus getInstance() {
        if (bus == null) {
            bus = new Bus();
        }
        return bus;
    }
}

MyEvent 自己定義的事件類

package com.example.net.mobctrl.ottotest;

public class MyEvent {
    private String content;

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

    public MyEvent(){

    }

    public MyEvent(String content) {
        super();
        this.content = content;
    }

}

運行結果

05-20 20:41:59.923: I/System.out(30320): debug:這是我產生的事件(@Produce)
05-20 20:41:59.923: I/System.out(30320): debug:onCreate
05-20 20:42:11.553: I/System.out(30320): debug:將我點擊的內容，發送出去

每次調用registe()方法是，會立即調用@Produce方法，將return的事件發送出去，由參數為MyEvent的@Subscribe方法接收並處理。bus.post()也是如此。

原理

主要是Bus.java裡面的代碼：
關鍵的方法有

public void register(Object object)
該方法的作用是查找object裡面所有帶有Produce和Subscribe注解的方法，並保存在Map中，並且會立即執行Produce注解的方法。

public void post(Object event)
發送事件event，根據之前注冊過的object裡面的方法，查找參數為event的Subscribe方法，並invoke該方法。這樣就達到了post之後，調用對應Subscribe方法的目的。

public void unregister(Object object)
注銷object，刪除掉map中保存的object的方法，釋放object，防止內存洩露。

Bus源代碼

具體代碼如下：

package com.squareup.otto;

import java.lang.reflect.InvocationTargetException;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
/*
 * @author Cliff Biffle
 * @author Jake Wharton
 */
public class Bus {
  public static final String DEFAULT_IDENTIFIER = default;

  /** All registered event handlers, indexed by event type. */
  private final ConcurrentMap, Set> handlersByType =
          new ConcurrentHashMap, Set>();

  /** All registered event producers, index by event type. */
  private final ConcurrentMap, EventProducer> producersByType =
          new ConcurrentHashMap, EventProducer>();

  /** Identifier used to differentiate the event bus instance. */
  private final String identifier;

  /** Thread enforcer for register, unregister, and posting events. */
  private final ThreadEnforcer enforcer;

  /** Used to find handler methods in register and unregister. */
  private final HandlerFinder handlerFinder;

  /** Queues of events for the current thread to dispatch. */
  private final ThreadLocal> eventsToDispatch =
      new ThreadLocal>() {
        @Override protected ConcurrentLinkedQueue initialValue() {
          return new ConcurrentLinkedQueue();
        }
      };

  /** True if the current thread is currently dispatching an event. */
  private final ThreadLocal isDispatching = new ThreadLocal() {
    @Override protected Boolean initialValue() {
      return false;
    }
  };

  /** Creates a new Bus named default that enforces actions on the main thread. */
  public Bus() {
    this(DEFAULT_IDENTIFIER);
  }

  /**
   * Creates a new Bus with the given {@code identifier} that enforces actions on the main thread.
   *
   * @param identifier a brief name for this bus, for debugging purposes.  Should be a valid Java identifier.
   */
  public Bus(String identifier) {
    this(ThreadEnforcer.MAIN, identifier);
  }

  /**
   * Creates a new Bus named default with the given {@code enforcer} for actions.
   *
   * @param enforcer Thread enforcer for register, unregister, and post actions.
   */
  public Bus(ThreadEnforcer enforcer) {
    this(enforcer, DEFAULT_IDENTIFIER);
  }

  /**
   * Creates a new Bus with the given {@code enforcer} for actions and the given {@code identifier}.
   *
   * @param enforcer Thread enforcer for register, unregister, and post actions.
   * @param identifier A brief name for this bus, for debugging purposes.  Should be a valid Java identifier.
   */
  public Bus(ThreadEnforcer enforcer, String identifier) {
    this(enforcer, identifier, HandlerFinder.ANNOTATED);
  }

  /**
   * Test constructor which allows replacing the default {@code HandlerFinder}.
   *
   * @param enforcer Thread enforcer for register, unregister, and post actions.
   * @param identifier A brief name for this bus, for debugging purposes.  Should be a valid Java identifier.
   * @param handlerFinder Used to discover event handlers and producers when registering/unregistering an object.
   */
  Bus(ThreadEnforcer enforcer, String identifier, HandlerFinder handlerFinder) {
    this.enforcer =  enforcer;
    this.identifier = identifier;
    this.handlerFinder = handlerFinder;
  }

  @Override public String toString() {
    return [Bus  + identifier + ];
  }

  /**
   * Registers all handler methods on {@code object} to receive events and producer methods to provide events.
   *

   * If any subscribers are registering for types which already have a producer they will be called immediately
   * with the result of calling that producer.
   *

   * If any producers are registering for types which already have subscribers, each subscriber will be called with
   * the value from the result of calling the producer.
   *
   * @param object object whose handler methods should be registered.
   * @throws NullPointerException if the object is null.
   */
  public void register(Object object) {
    if (object == null) {
      throw new NullPointerException(Object to register must not be null.);
    }
    enforcer.enforce(this);

    Map, EventProducer> foundProducers = handlerFinder.findAllProducers(object);
    for (Class type : foundProducers.keySet()) {

      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() + .);
      }
      Set handlers = handlersByType.get(type);
      if (handlers != null && !handlers.isEmpty()) {
        for (EventHandler handler : handlers) {
          dispatchProducerResultToHandler(handler, producer);
        }
      }
    }

    Map, Set> foundHandlersMap = handlerFinder.findAllSubscribers(object);
    for (Class type : foundHandlersMap.keySet()) {
      Set handlers = handlersByType.get(type);
      if (handlers == null) {
        //concurrent put if absent
        Set handlersCreation = new CopyOnWriteArraySet();
        handlers = handlersByType.putIfAbsent(type, handlersCreation);
        if (handlers == null) {
            handlers = handlersCreation;
        }
      }
      final Set foundHandlers = foundHandlersMap.get(type);
      if (!handlers.addAll(foundHandlers)) {
        throw new IllegalArgumentException(Object already registered.);
      }
    }

    for (Map.Entry, Set> entry : foundHandlersMap.entrySet()) {
      Class type = entry.getKey();
      EventProducer producer = producersByType.get(type);
      if (producer != null && producer.isValid()) {
        Set foundHandlers = entry.getValue();
        for (EventHandler foundHandler : foundHandlers) {
          if (!producer.isValid()) {
            break;
          }
          if (foundHandler.isValid()) {
            dispatchProducerResultToHandler(foundHandler, producer);
          }
        }
      }
    }
  }

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

  /**
   * Unregisters all producer and handler methods on a registered {@code object}.
   *
   * @param object object whose producer and handler methods should be unregistered.
   * @throws IllegalArgumentException if the object was not previously registered.
   * @throws NullPointerException if the object is null.
   */
  public void unregister(Object object) {
    if (object == null) {
      throw new NullPointerException(Object to unregister must not be null.);
    }
    enforcer.enforce(this);

    Map, EventProducer> producersInListener = handlerFinder.findAllProducers(object);
    for (Map.Entry, 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();
    }

    Map, Set> handlersInListener = handlerFinder.findAllSubscribers(object);
    for (Map.Entry, Set> entry : handlersInListener.entrySet()) {
      Set currentHandlers = getHandlersForEventType(entry.getKey());
      Collection 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);
    }
  }

  /**
   * Posts an event to all registered handlers.  This method will return successfully after the event has been posted to
   * all handlers, and regardless of any exceptions thrown by handlers.
   *
   * 

If no handlers have been subscribed for {@code event}'s class, and {@code event} is not already a * {@link DeadEvent}, it will be wrapped in a DeadEvent and reposted. * * @param event event to post. * @throws NullPointerException if the event is null. */ public void post(Object event) { if (event == null) { throw new NullPointerException(Event to post must not be null.); } enforcer.enforce(this); Set> dispatchTypes = flattenHierarchy(event.getClass()); boolean dispatched = false; for (ClasseventType : dispatchTypes) { Set wrappers = getHandlersForEventType(eventType); if (wrappers != null && !wrappers.isEmpty()) { dispatched = true; for (EventHandler wrapper : wrappers) { enqueueEvent(event, wrapper); } } } if (!dispatched && !(event instanceof DeadEvent)) { post(new DeadEvent(this, event)); } dispatchQueuedEvents(); } /** * Queue the {@code event} for dispatch during {@link #dispatchQueuedEvents()}. Events are queued in-order of * occurrence so they can be dispatched in the same order. */ protected void enqueueEvent(Object event, EventHandler handler) { eventsToDispatch.get().offer(new EventWithHandler(event, handler)); } /** * Drain the queue of events to be dispatched. As the queue is being drained, new events may be posted to the end of * the queue. */ 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. if (isDispatching.get()) { return; } isDispatching.set(true); try { while (true) { EventWithHandler eventWithHandler = eventsToDispatch.get().poll(); if (eventWithHandler == null) { break; } if (eventWithHandler.handler.isValid()) { dispatch(eventWithHandler.event, eventWithHandler.handler); } } } finally { isDispatching.set(false); } } /** * Dispatches {@code event} to the handler in {@code wrapper}. This method is an appropriate override point for * subclasses that wish to make event delivery asynchronous. * * @param event event to dispatch. * @param wrapper wrapper that will call the 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); } } /** * Retrieves the currently registered producer for {@code type}. If no producer is currently registered for * {@code type}, this method will return {@code null}. * * @param type type of producer to retrieve. * @return currently registered producer, or {@code null}. */ EventProducer getProducerForEventType(Classtype) { return producersByType.get(type); } /** * Retrieves a mutable set of the currently registered handlers for {@code type}. If no handlers are currently * registered for {@code type}, this method may either return {@code null} or an empty set. * * @param type type of handlers to retrieve. * @return currently registered handlers, or {@code null}. */ Set getHandlersForEventType(Classtype) { return handlersByType.get(type); } /** * Flattens a class's type hierarchy into a set of Class objects. The set will include all superclasses * (transitively), and all interfaces implemented by these superclasses. * * @param concreteClass class whose type hierarchy will be retrieved. * @return {@code concreteClass}'s complete type hierarchy, flattened and uniqued. */ Set> flattenHierarchy(ClassconcreteClass) { Set> classes = flattenHierarchyCache.get(concreteClass); if (classes == null) { classes = getClassesFor(concreteClass); flattenHierarchyCache.put(concreteClass, classes); } return classes; } private Set> getClassesFor(ClassconcreteClass) { List> parents = new LinkedList>(); Set> classes = new HashSet>(); parents.add(concreteClass); while (!parents.isEmpty()) { Classclazz = parents.remove(0); classes.add(clazz); Classparent = clazz.getSuperclass(); if (parent != null) { parents.add(parent); } } return classes; } /** * Throw a {@link RuntimeException} with given message and cause lifted from an {@link * InvocationTargetException}. If the specified {@link InvocationTargetException} does not have a * cause, neither will the {@link RuntimeException}. */ private static void throwRuntimeException(String msg, InvocationTargetException e) { Throwable cause = e.getCause(); if (cause != null) { throw new RuntimeException(msg + : + cause.getMessage(), cause); } else { throw new RuntimeException(msg + : + e.getMessage(), e); } } private final Map, Set>> flattenHierarchyCache = new HashMap, Set>>(); /** Simple struct representing an event and its handler. */ static class EventWithHandler { final Object event; final EventHandler handler; public EventWithHandler(Object event, EventHandler handler) { this.event = event; this.handler = handler; } } }

有趣的小工具AnnotatedHandlerFinder

當你自己寫框架的時候，很多時候需要用到Annotation查找，

package com.squareup.otto;

import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * Helper methods for finding methods annotated with {@link Produce} and {@link Subscribe}.
 *
 * @author Cliff Biffle
 * @author Louis Wasserman
 * @author Jake Wharton
 */
final class AnnotatedHandlerFinder {

  /** Cache event bus producer methods for each class. */
  private static final Map, Map, Method>> PRODUCERS_CACHE =
      new HashMap, Map, Method>>();

  /** Cache event bus subscriber methods for each class. */
  private static final Map, Map, Set>> SUBSCRIBERS_CACHE =
      new HashMap, Map, Set>>();

  /**
   * Load all methods annotated with {@link Produce} or {@link Subscribe} into their respective caches for the
   * specified class.
   */
  private static void loadAnnotatedMethods(Class listenerClass) {
    Map, Set> subscriberMethods = new HashMap, Set>();
    Map, Method> producerMethods = new HashMap, Method>();

    for (Method method : listenerClass.getDeclaredMethods()) {
      // The compiler sometimes creates synthetic bridge methods as part of the
      // type erasure process. As of JDK8 these methods now include the same
      // annotations as the original declarations. They should be ignored for
      // subscribe/produce.
      if (method.isBridge()) {
        continue;
      }
      if (method.isAnnotationPresent(Subscribe.class)) {
        Class[] parameterTypes = method.getParameterTypes();
        if (parameterTypes.length != 1) {
          throw new IllegalArgumentException(Method  + method +  has @Subscribe annotation but requires 
              + parameterTypes.length +  arguments.  Methods must require a single argument.);
        }

        Class eventType = parameterTypes[0];
        if (eventType.isInterface()) {
          throw new IllegalArgumentException(Method  + method +  has @Subscribe annotation on  + eventType
              +  which is an interface.  Subscription must be on a concrete class type.);
        }

        if ((method.getModifiers() & Modifier.PUBLIC) == 0) {
          throw new IllegalArgumentException(Method  + method +  has @Subscribe annotation on  + eventType
              +  but is not 'public'.);
        }

        Set methods = subscriberMethods.get(eventType);
        if (methods == null) {
          methods = new HashSet();
          subscriberMethods.put(eventType, methods);
        }
        methods.add(method);
      } else if (method.isAnnotationPresent(Produce.class)) {
        Class[] parameterTypes = method.getParameterTypes();
        if (parameterTypes.length != 0) {
          throw new IllegalArgumentException(Method  + method + has @Produce annotation but requires 
              + parameterTypes.length +  arguments.  Methods must require zero arguments.);
        }
        if (method.getReturnType() == Void.class) {
          throw new IllegalArgumentException(Method  + method
              +  has a return type of void.  Must declare a non-void type.);
        }

        Class eventType = method.getReturnType();
        if (eventType.isInterface()) {
          throw new IllegalArgumentException(Method  + method +  has @Produce annotation on  + eventType
              +  which is an interface.  Producers must return a concrete class type.);
        }
        if (eventType.equals(Void.TYPE)) {
          throw new IllegalArgumentException(Method  + method +  has @Produce annotation but has no return type.);
        }

        if ((method.getModifiers() & Modifier.PUBLIC) == 0) {
          throw new IllegalArgumentException(Method  + method +  has @Produce annotation on  + eventType
              +  but is not 'public'.);
        }

        if (producerMethods.containsKey(eventType)) {
          throw new IllegalArgumentException(Producer for type  + eventType +  has already been registered.);
        }
        producerMethods.put(eventType, method);
      }
    }

    PRODUCERS_CACHE.put(listenerClass, producerMethods);
    SUBSCRIBERS_CACHE.put(listenerClass, subscriberMethods);
  }

  /** This implementation finds all methods marked with a {@link Produce} annotation. */
  static Map, EventProducer> findAllProducers(Object listener) {
    final Class listenerClass = listener.getClass();
    Map, EventProducer> handlersInMethod = new HashMap, EventProducer>();

    if (!PRODUCERS_CACHE.containsKey(listenerClass)) {
      loadAnnotatedMethods(listenerClass);
    }
    Map, Method> methods = PRODUCERS_CACHE.get(listenerClass);
    if (!methods.isEmpty()) {
      for (Map.Entry, Method> e : methods.entrySet()) {
        EventProducer producer = new EventProducer(listener, e.getValue());
        handlersInMethod.put(e.getKey(), producer);
      }
    }

    return handlersInMethod;
  }

  /** This implementation finds all methods marked with a {@link Subscribe} annotation. */
  static Map, Set> findAllSubscribers(Object listener) {
    Class listenerClass = listener.getClass();
    Map, Set> handlersInMethod = new HashMap, Set>();

    if (!SUBSCRIBERS_CACHE.containsKey(listenerClass)) {
      loadAnnotatedMethods(listenerClass);
    }
    Map, Set> methods = SUBSCRIBERS_CACHE.get(listenerClass);
    if (!methods.isEmpty()) {
      for (Map.Entry, Set> e : methods.entrySet()) {
        Set handlers = new HashSet();
        for (Method m : e.getValue()) {
          handlers.add(new EventHandler(listener, m));
        }
        handlersInMethod.put(e.getKey(), handlers);
      }
    }

    return handlersInMethod;
  }

  private AnnotatedHandlerFinder() {
    // No instances.
  }

}