RxBus学习之旅－－从入门到提高

在公司的技术分享会上，做了关于RxBus的学习分享，记录如下：

一.RxBus与RxJava

一次RxJava调用过程可以划分为以下环节：

创建观察内容 （片段1）数据处理／映射（片段2）选择线程（片段3）订阅（片段4，片段5）完成／错误处理（片段6）

示例代码：

Observable // 片段1 .create(new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { subscriber.onStart(); String trim = mainEd.getText().toString().trim(); subscriber.onNext(trim); subscriber.onError(new Throwable()); subscriber.onCompleted(); } }) // 片段2 .map(new Func1<String, String>() { @Override public String call(String s) { return s + " sugarya"; } }) // 片段3 .subscribeOn(Schedulers.newThread()) .observeOn(AndroidSchedulers.mainThread()) // 片段4 .subscribe( // 片段5 new Subscriber<String>() { @Override public void onStart() { super.onStart(); } //片段6 @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(String s) { mainTv2.setText(s); } });

二.RxBus与接口回调

一次完整的接口回调，包括四个步骤：

接口定义接口调用接口实现接口注入

RxBus的使用过程，就是一个接口回调的过程。

接口定义，在RxJava定义好了。

上面示例代码片段1里的suscriber.onNext(),onStart(),onError(),onComplete()对应接口调用。

代码片段5，片段6这些是接口实现

注入的过程是调用suscribe()方法订阅的过程

三.RxBus源码分析

RxBus的代码实现如下：

public class RxBus { private static volatile RxBus instance; private final Subject<Object, Object> _bus; private RxBus() { _bus = new SerializedSubject<>(PublishSubject.create()); } public static RxBus getInstance() { if (null == instance) { synchronized (RxBus.class) { if (null == instance) { instance = new RxBus(); } } } return instance; } public void send(Object object) { try{ _bus.onNext(object); }catch (Exception e){ e.printStackTrace(); } } public boolean hasObservers() { return _bus.hasObservers(); } private <T> Observable<T> toObservable(final Class<T> type) { return _bus.ofType(type);//filter + cast } public <T> Subscription toSubscription(final Class<T> type, Observer<T> observer) { return toObservable(type).subscribe(observer); } public <T> Subscription toSubscription(final Class<T> type, Action1<T> action1) { return toObservable(type).subscribe(action1); } public <T> Subscription toSubscription(final Class<T> type, Action1<T> action1, Action1<Throwable> errorAction1) { return toObservable(type).subscribe(action1,errorAction1); } }

接下来对上述代码做些简要分析：

volatile

保证instance可见性禁止指令重排

这里涉及到Java内存模型，相关资料： 传送门

SerializedSubject

SerializedSubject extends Subject extends Observable implements Observer，既是观察内容，又是观察者，起到桥梁／数据转发的作用

保证多线程安全，Subject 当作一个 Subscriber 使用，从多个线程中调用它的onNext方法（包括其它的on系列方法）

PublishSubject

主题，RxJava里有四种主题

PublishSubjectBehaviorSubject ReplaySubjectAsyncSubject

PublishSubject的含义是：在订阅者订阅的时间点之后的数据发送给观察者

ofType操作符 = filter操作符 + cast操作符

filter只有符合过滤条件的数据才会被“发射”cast将一个Observable转换成指定类型的Observable

CompositeSubscription

该对象作为subscription的容器，方便统一取消订阅

四.RxBus异常处理

当RxBus在执行过程中，任意环节发生了错误异常，订阅关系就会被取消。之后再次发送，将无法执行订阅后的回调。

做了一个数组越界的错误来演示

@Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main); ButterKnife.bind(this); onRxBus(); } private void onRxBus() { int[] array = new int[2]; RxBus.getInstance().toSubscription(Integer.class, new Subscriber<Integer>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { //onRxBus(); Log.e(TAG, "mSubscriber17 onError: " + e.toString()); } @Override public void onNext(Integer integer) { array[integer] = integer; Log.e(TAG, "mSubscriber17 onNext:" + integer); } }); } @OnClick(R.id.btn_main17) void onClick17() { RxBus.getInstance().send(2); }

这时候点击button按钮，数组越界异常，第二次再点击Button发送消息，没有响应了。这里，RxBus，确切的说是RxJava捕获到错误异常，就会取消订阅关系。

E/MainActivity: mSubscriber17 onError: java.lang.ArrayIndexOutOfBoundsException: length=2; index=2

解决的思路：

使用try－catch捕获异常，不让异常被RxJava捕获在onError里重新订阅。

接下来说说第二种方法，具体怎么操作，其实就是在onError方法里，重新执行一遍订阅，执行上述注释掉的代码onRxBus,就解决问题了。

五.小结

上述其实是这次技术分享大纲，技术分享准备功课前，刷了下面的文章，要对RxBus有更细致的学习和了解，可以阅读：（推荐）

谢晨成系列

RxBus简单实现

RxBus深入源码解析

RxJava里onError异常处理

Yokey系列

RxJava实现事件总线

RxBus异常处理

RxBus实现Sticky事件（粘性订阅）

其他

RxBus从基础实现到升级——打造属于自己的RxBus

EventBus和RxBus实现和性能比较