Android(安卓)积累一些RxJava2.0有关的知识

最近偶然逛逛“淘宝（找代码）”,发现rxjava变rxjava2了。

变化1上码：

Observable.create(new ObservableOnSubscribe() {            @Override            public void subscribe(ObservableEmitter e) throws Exception {                e.onNext("1");                e.onNext("2");                e.onNext("3");                e.onNext("4");                e.onComplete();                e.onNext("5");            }        }).subscribe(new Observer() {            @Override            public void onSubscribe(Disposable d) {                dis=d;                Log.i("Rxjava2::onSubscribe","d");            }            @Override            public void onNext(String o) {                Log.i("Rxjava2::onNext",o);                if (o.equals("2")){                    dis.dispose();                }                tv.setText(o);            }            @Override            public void onError(Throwable e) {                Log.i("Rxjava2::onError",e.getMessage().toString());            }            @Override            public void onComplete() {                Log.i("Rxjava2::onComplete","c");            }        });

貌似多了：

ObservableEmitter:用于向观察者发送数据。（我觉得这里可以理解成服务器【被观察者】与客户端【观察者】的关系，被观察者向观察者发送onnext(),onerror,oncomplete三种指令，只不过onnext可携带参数），这里明确的是subcrible里的代码坑定会执行的，无论观察者愿不愿意收到，只要绑定了，就会发，只不过观察者可以选择不接受。

Disposable:被观察者向观察者发送数据，如果观察者不想接怎么办，就得调用dispose()方法，当然被观察者也可以打断观察者接受数据，只要调用oncomplete或者onerror方法，这一点和rxjava1.x是一样的

变化2：貌似没有什么action1 function1 之类的回调了。

番外:个人用rxjava1.x的开发的时候就贼喜欢用action1的回调其他的一概不写，认为没必要。然而血的教训就是遇到这个bug:

java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. Add `onError` handling.

Caused by: rx.exceptions.OnErrorNotImplementedException

Caused by: rx.exceptions.MissingBackpressureException

大概原因是：少了onerror的回调处理（场景还原：屏幕按黑了再过一会按亮，就蹦）【个人观点错的】

坑爹指数：尼玛拉下来的log一看，不报哪一行出错

解决办法：用rxjava的地方，用lamb加上onerror处理？（貌似没再崩锅）

当然我的想法绝逼是错的，在读过这位大神的著作之后：

把玩他吧

上码：

        Observable.just("ddd").subscribe(new Consumer() {            @Override            public void accept(@NonNull String s) throws Exception {            }        });

貌似加了个抛出异常，第一眼觉得很6（操蛋）。

  Observable.just("ddd").subscribe((msg)->{Log.i("接收的值",msg);},Throwable::printStackTrace);

或者像rxjava1一样写？

  Observable.just("ddd").subscribe((msg)->{Log.i("接收的值",msg);},(error)->{Log.i("error",error.getMessage());});

Retrofit2+Rxjava2的变化

        Retrofit adapter = new Retrofit.Builder()                .baseUrl(endpoint)                .addConverterFactory(ScalarsConverterFactory.create())                .addConverterFactory(GsonConverterFactory.create())                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())                .build();        return adapter.create(serviceClass);

//用rxjava2轮询，轮询有限次数，每隔0.1s轮询一次，总共轮询20次。

 Flowable.interval(0, 100, TimeUnit.MILLISECONDS)                .take(20)                .subscribeOn(Schedulers.io())                .subscribe(new Subscriber() {                    @Override                    public void onSubscribe(Subscription s) {                        s.request(100);                        mCheckUpdate = Disposables.fromSubscription(s);                    }                    @Override                    public void onNext(Long aLong) {                        if (isNetworkAvailable()) {                            processor.onNext("checkUpdate");                            mCheckUpdate.dispose();                        }                    }                    @Override                    public void onError(Throwable t) {                    }                    @Override                    public void onComplete() {//                        getAssociated();                        showDialog();                    }                });

更多相关文章

随机推荐