RxJava：RxBus事件总线

Photo by Peter Hershey on Unsplash

学习于：http://www.jianshu.com/p/ca090f6e2fe2
http://reactivex.io/documentation/subject.html

基于RxJava2，通过观察者模式实现页面之间的数据传递；

public class RxBus {

    private static volatile RxBus mInstance;
    private final Subject<Object> bus;
    public RxBus() {
        bus = PublishSubject.create().toSerialized();

    }
    // 单例RxBus
    public static RxBus getDefault() {
        if (mInstance == null) {
            synchronized (RxBus.class) {
                if (mInstance == null) {
                    mInstance = new RxBus();
                }
            }
        }
        return mInstance;
    }
    // 发送事件
    public void post(Object o) {
        bus.onNext(o);
    }
    public <T> Observable<T> toObservable(Class<T> eventType) {
        return bus.ofType(eventType);
    }
}

Subject：既作为Observer又作为Observable，并且提供了4种Subject，
此处我们仅仅是简单的实现RxBus，所以使用PublishSubject；
PublishSubject发送订阅之后的全部数据；

相同的是如果源Observable以错误终止，Subject则不会向随后的观察者发出任何项目，但会将简单地传递源Observable中的错误通知。

Subject是非线程安全的，为了在并发情况下只一个线程调用onNext()等方法，所以通过toSerialized()方法将Subject对象转换为SerializedSubject对象；

发送事件

    public void btn_2(View view) {
        RxBus.getDefault().post(new RxBusPostBean("whstywh", "18"));
    }

事件类

public class RxBusPostBean {
    private String name;
    private String age;

    public RxBusPostBean(String name, String age) {
        this.name = name;
        this.age = age;
    }

    public String getName() {
        return name;
    }
    
    public String getAge() {
        return age;
    }
}

接受事件

    private void getDataByRxBus() {
        mDisposable = RxBus.getDefault()
                .toObservable(RxBusPostBean.class)
                .subscribe(new Consumer<RxBusPostBean>() {
                    @Override
                    public void accept(RxBusPostBean rxBusPostBean) throws Exception {
                        mTextView1.setText(rxBusPostBean.getName());
                        mTextView2.setText(rxBusPostBean.getAge());
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.d("flag", "throwable:" + throwable.getMessage());
                    }
                });
    }

在生命周期结束的地方取消订阅事件，防止RxJava可能会引起的内存泄漏问题

    @Override
    public void onDestroy() {
        super.onDestroy();
        if (!mDisposable.isDisposed()) {
            mDisposable.dispose();
        }
    }

对于RxJava的学习才刚刚开始！
RxJava：RxBus事件总线
RxJava：RxBinding
RxJava：Rxlifecycle