Rxbus实现事件总线（RxJava2.0）

封装base的时候想到事件分发 既然用到了Rxjava 很神奇的发现了Rxbus这个东西，那就学习下~
Github:https://github.com/balckban/RxbusDemo-RxJava2.0-

ezgif-1-3b2413c862.gif

1. 首先是gradle的配置

compile "io.reactivex.rxjava2:rxjava:2.1.3"
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

2. RxBus写法
   背压式处理 很详细http://flyou.ren/2017/04/05/关于RxJava背压/
   1.没有做背压式处理

private final Subject<Object> _bus;


    private static class RxHolder{
        private static volatile RxBus sRxBus = new RxBus();
    }

    private RxBus() {
        _bus = PublishSubject.create().toSerialized();
    }
    public static RxBus getInstance(){
        return RxHolder.sRxBus;
    }
    public void send(Object o){
        if (hasObserve()) {
            _bus.onNext(o);
        }
    }

    private boolean hasObserve() {
        return _bus.hasObservers();
    }
    public Observable<Object> toObservable(){
        return _bus;
    }
    // 根据传入的事件类型,返回特定类型的eventType的被观察者
    public <T extends Object> Observable<T> toObservable(Class<T> event){
        return _bus.ofType(event);

    }

2.背压式处理代码

   private final FlowableProcessor<Object> _bus;

    private RxBus() {
        // toSerialized method made bus thread safe
        _bus = PublishProcessor.create().toSerialized();
    }

    public static RxBus getInstance() {
        return Holder.BUS;
    }

    public void send(Object obj) {
        _bus.onNext(obj);
    }
/**
 * 根据传递的 eventType 类型返回特定事件类型的被观察者
 */
    public <T> Flowable<T> toFlowable(Class<T> tClass) {
        return _bus.ofType(tClass);
    }

    public Flowable<Object> toFlowable() {
        return _bus;
    }

    public boolean hasSubscribers() {
        return _bus.hasSubscribers();
    }

    private static class Holder {
        private static final RxBus BUS = new RxBus();
    }

3.Rxbus使用
接收

        RxBus.getInstance().toFlowable(输入自己特定的class 比如TestEvent.class 区分事件).subscribe(new Consumer<Object>() {

            @Override
            public void accept(Object o) throws Exception {
                Toast.makeText(MainActivity.this, "已发送", Toast.LENGTH_SHORT).show();
                if (((TestEvent) o).getTest() != null) {
                    mTvReceive.setText(((TestEvent) o).getTest());
                }
            }
        });

    }

   

发送

mBtnReturn.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                RxBus.getInstance().send(new TestEvent(mEtNext.getText().toString()));
                finish();
            }
        });