Android事件通知工具：RxBus在Eclipse和AS中的实践

/**
 * RxBus
 * 
 * @author xx
 * 
 */
public class RxBus {
  private ConcurrentHashMap<Object, List> maps = new ConcurrentHashMap<Object, List>();
  private static RxBus sInstance;
  public static RxBus get() {
    if (sInstance == null) {
      synchronized (RxBus.class) {
        if (sInstance == null) {
          sInstance = new RxBus();
        }
      }
    }
    return sInstance;
  }
  /**
   * 简单以对象的类名注册
   * @param o
   * @return
   */
  public  Observable simpleRegister(@NonNull Object o) {
    return register(o.getClass().getSimpleName());
  }
  /**
   * 注册订阅
   * @param tag
   * @return
   */
  public  Observable register(@NonNull Object tag) {
    List subjects = maps.get(tag);
    if (subjects == null) {
      subjects = new ArrayList();
      maps.put(tag, subjects);
    }
    Subject subject = PublishSubject. create();
    subjects.add(subject);
    return subject;
  }
  /**
   * 注销tag制定的订阅
   * @param tag
   */
  public void unregister(@NonNull Object tag, @NonNull Observable observable) {
    List subjects = maps.get(tag);
    if (subjects != null) {
      subjects.remove((Subject) observable);
      if (subjects.isEmpty()) {
        maps.remove(tag);
      }
    }
  }
  /**
   * 注销tag所有的订阅
   * @param tag
   */
  public void unregisterAll(@NonNull Object tag) {
    List subjects = maps.get(tag);
    if (subjects != null) {
      maps.remove(tag);
    }
  }
  public void post(@NonNull Object tag) {
    post(tag, tag);
  }
  /**
   * 发送指定tag的事件
   * @param tag
   * @param o
   */
  @SuppressWarnings("unchecked")
  public void post(@NonNull Object tag, @NonNull Object o) {
    List subjects = maps.get(tag);
    if (subjects != null && !subjects.isEmpty()) {
      for (Subject s : subjects) {
        s.onNext(o);
      }
    }
  }
}
public class RxManager {
  private final static String TAG = "RxManager";
  private static RxManager sInstance;
  private RxBus mRxBus = RxBus.get();
  /** 管理Subscribers订阅，防止内存泄漏 */
  private ConcurrentHashMap maps = new ConcurrentHashMap();
  public static RxManager get() {
    if (sInstance == null) {
      synchronized (RxManager.class) {
        if (sInstance == null) {
          sInstance = new RxManager();
        }
      }
    }
    return sInstance;
  }
  /**
   * 
   * RxBus注入监听
   * 
   * @param eventName
   * 
   * @param action1
   */
  public  void onMainThread(Object eventName, Action1 action1) {
    Observable Observable = mRxBus.register(eventName);
    /* 订阅管理 */
    add(eventName, Observable.observeOn(AndroidSchedulers.mainThread()).subscribe(action1, new Action1() {
      @Override
      public void call(Throwable throwable) {
        Log.e(TAG, throwable.getMessage());
      }
    }));
  }
  /**
   * 
   * RxBus注入监听
   * 
   * @param eventName
   *            事件名
   * @param action1
   */
  public  void on(Object eventName, Action1 action1) {
    Observable Observable = mRxBus.register(eventName);
    /* 订阅管理 */
    add(eventName, Observable.subscribe(action1, new Action1() {
      @Override
      public void call(Throwable throwable) {
        Log.e(TAG, throwable.getMessage());
      }
    }));
  }
  /**
   * 单纯的Observables 和Subscribers管理
   * 
   * @param m
   */
  public void add(Object eventName, Subscription m) {
    /* 订阅管理 */
    CompositeSubscription subscription = maps.get(eventName);
    if (subscription == null) {
      subscription = new CompositeSubscription();
      maps.put(eventName, subscription);
    }
    subscription.add(m);
  }
  /**
   * 
   * 单个presenter生命周期结束，取消订阅和所有rxbus观察
   */
  public void clear(@NonNull Object eventName) {
    CompositeSubscription subscription = maps.get(eventName);
    if (subscription != null) {
      subscription.unsubscribe(); // 取消订阅
      maps.remove(eventName);
    }
    mRxBus.unregisterAll(eventName);
  }
  // 发送rxbus
  public void post(Object tag) {
    mRxBus.post(tag);
  }
  // 发送rxbus
  public void post(Object tag, Object content) {
    mRxBus.post(tag, content);
  }
}