详解用RxJava实现事件总线(Event Bus)

/**
* RxBus
* Created by YoKeyword on 2015/6/17.
*/
public class RxBus {
  private static volatile RxBus defaultInstance;

  private final Subject<Object, Object> bus;
  // PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者
  public RxBus() {
   bus = new SerializedSubject<>(PublishSubject.create());
  }
  // 单例RxBus
  public static RxBus getDefault() {
    if (defaultInstance == null) {
      synchronized (RxBus.class) {
        if (defaultInstance == null) {
          defaultInstance = new RxBus();
        }
      }
    }
    return defaultInstance ;
  }
  // 发送一个新的事件
  public void post (Object o) {
    bus.onNext(o);
  }
  // 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者
  public <T> Observable<T> toObservable (Class<T> eventType) {
    return bus.ofType(eventType);
//    这里感谢小鄧子的提醒: ofType = filter + cast
//    return bus.filter(new Func1<Object, Boolean>() {
//      @Override
//      public Boolean call(Object o) {
//        return eventType.isInstance(o);
//      }
//    }) .cast(eventType);
  }
}

public final <R> Observable<R> ofType(final Class<R> klass) {
  return filter(new Func1<T, Boolean>() {
    @Override
    public final Boolean call(T t) {
      return klass.isInstance(t);
    }
  }).cast(klass);
}

RxBus.getDefault().post(new UserEvent (1, "yoyo"));

public class UserEvent {
  long id;
  String name;
  public UserEvent(long id,String name) {
    this.id= id;
    this.name= name;
  }
  public long getId() {
    return id;
  }
  public String getName() {
    return name;
  }
}

// rxSubscription是一个Subscription的全局变量，这段代码可以在onCreate/onStart等生命周期内
rxSubscription = RxBus.getDefault().toObserverable(UserEvent.class)
    .subscribe(new Action1<UserEvent>() {
        @Override
        public void call(UserEvent userEvent) {
          long id = userEvent.getId();
          String name = userEvent.getName();
          ...
        }
      },
    new Action1<Throwable>() {
      @Override
      public void call(Throwable throwable) {
        // TODO: 处理异常
      }    
    });

@Override
protected void onDestroy() {
  super.onDestroy();
  if(!rxSubscription.isUnsubscribed()) {
    rxSubscription.unsubscribe();
  }
}