为什么要匿名？
在计算机程序设计中有一个非常棒的思想叫“解耦”。你通常希望在你的设计中保持尽可能低的耦合度。
通常情况下，你希望消息发布商能够直接了解所有需要接收消息的订阅者，这样，一旦“事件”或消息准备好就可以及时通知每一个订阅者。但是使用事件总线，发布者可以免除这种职责并实现独立性，因为消息发布者和消息订阅者可以相互不知道对方，只关心对应的消息，从而接触两者之间的依赖关系

怎么实现匿名？
提到匿名，自然而然你就会问：你是如何真正实现发布者和订阅者之间的匿名？ 很简单，只要找到一个中间人，让这个中间人负责两方的消息调度。事件总线就是一个这样的中间人。综上所述，事件总线就是这么简单。

public class RxBus {private final Subject<Object, Object> _bus = new SerializedSubject<>(PublishSubject.create());public void send(Object o) {_bus.onNext(o);}public Observable<Object> toObserverable() {return _bus;}
}

@OnClick(R.id.btn_demo_rxbus_tap)
public void onTapButtonClicked() {_rxBus.send(new TapEvent());
}

_rxBus.toObserverable().subscribe(new Action1<Object>() {@Overridepublic void call(Object event) {if(event instanceof TapEvent) {_showTapText();}else if(event instanceof SomeOtherEvent) {_doSomethingElse();}}});

/*** desc   : 利用 PublishSubject的特性：与普通的Subject不同，在订阅时并不立即触发订阅事件，* 而是允许我们在任意时刻手动调用onNext(),onError(),onCompleted来触发事件。*/
public class RxBus {private ConcurrentHashMap<Object, List<Subject>> subjectMapper = new ConcurrentHashMap<>();private RxBus() {}private static class Holder {private static RxBus instance = new RxBus();}public static RxBus getInstance() {return Holder.instance;}public <T> Observable<T> register(@NonNull Class<T> clz) {return register(clz.getName());}public <T> Observable<T> register(@NonNull Object tag) {List<Subject> subjectList = subjectMapper.get(tag);if (null == subjectList) {subjectList = new ArrayList<>();subjectMapper.put(tag, subjectList);}Subject<T, T> subject = PublishSubject.create();subjectList.add(subject);//System.out.println("注册到rxbus");return subject;}public <T> void unregister(@NonNull Class<T> clz, @NonNull Observable observable) {unregister(clz.getName(), observable);}public void unregister(@NonNull Object tag, @NonNull Observable observable) {List<Subject> subjects = subjectMapper.get(tag);if (null != subjects) {subjects.remove(observable);if (subjects.isEmpty()) {subjectMapper.remove(tag);//System.out.println("从rxbus取消注册");
            }}}public void post(@NonNull Object content) {post(content.getClass().getName(), content);}public void post(@NonNull Object tag, @NonNull Object content) {List<Subject> subjects = subjectMapper.get(tag);if (!subjects.isEmpty()) {for (Subject subject: subjects) {subject.onNext(content);}}}
}

几个关键方法：
register —— 由tag，生成一个subject List，同时利用PublishSubject创建一个Subject并返回，它同时也是Observable的子类。
unregister —— 移除tag对应subject List 中的Observable。若subject List为空，也将被移除。
post —— 遍历tag对应subject List 中的Subject，执行onNext()。这里实际执行的是观察者Observer的onNext()，Subject的定义：public abstract class Subject<T, R> extends Observable<R> implements Observer<T>。

/*
rxbus*/
Observable<String> observable = RxBus.getInstance().register(String.class);
observable.map(s -> {try {int v = Integer.valueOf(s);System.out.println("map变换成功, source = " + s);return v;} catch (Exception e) {System.out.println("map变换失败, source = " + s);return s;}
}).subscribe(value -> {System.out.println("订阅 " + value);
});RxBus.getInstance().post("888");
RxBus.getInstance().post("发发发");
RxBus.getInstance().unregister(String.class, observable);
//这里比较有意思的是，使用了lambda表达式。//在map变换时，如果将字符串转成Integer，没有问题就返回整型；//若报异常，就返回String型。//同样的，在最终订阅时，value参数的类型也是由map变换来决定的。

public class RxBus {private final FlowableProcessor<Object> mBus;private RxBus() {mBus = PublishProcessor.create().toSerialized();}private static class Holder {private static RxBus instance = new RxBus();}public static RxBus getInstance() {return Holder.instance;}public void post(@NonNull Object obj) {mBus.onNext(obj);}public <T> Flowable<T> register(Class<T> clz) {return mBus.ofType(clz);}public void unregisterAll() {//会将所有由mBus 生成的 Flowable 都置  completed 状态  后续的 所有消息  都收不到了
        mBus.onComplete();}public boolean hasSubscribers() {return mBus.hasSubscribers();}}

Flowable<Integer> f1 = RxBus.getInstance().register(Integer.class);
f1.subscribe(value -> System.out.println("订阅f1消息 .. " + value));
RxBus.getInstance().post(999);

 @Overridepublic void fetchData() {observable = RxBus.getInstance().register(BaseListFragment.TAG);observable.subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {adapter.notifyDataSetChanged();}});}

 Disposable disposable = observable.subscribe(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) throws Exception {//这里接收数据项
            }}, new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) throws Exception {//这里接收onError
            }}, new Action() {@Overridepublic void run() throws Exception {//这里接收onComplete。
            }});