对于 RxJava2 的 认知与直接应用（一）

最近有时间学习些许内容，加上重新写博客来记录自己的学习过程与心得

1.rxjava2 认知

rxjava作为知名的响应式编程库，这半年内极大的火爆开发者中 介绍 Observable 被观察者 | | subscribe() 订阅 | | Observer 观察者

依赖： compile ‘io.reactivex.rxjava2:rxjava:2.0.5’ compile ‘io.reactivex.rxjava2:rxandroid:2.0.1’

Observable写法：以发送integer为例子

Observable<Integer> observable=Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter<Integer> e) throws Exception { e.onNext(1); e.onNext(2); e.onNext(3); e.onComplete(); } });

Observer 写法：

Observer<Integer> observer=new Observer<Integer>() { //自动重写以下方法 @Override public void onSubscribe(Disposable d) { Log.i(TAG, "onSubscribe: "); } @Override public void onNext(Integer integer) { Log.i(TAG, "onNext: "+integer); } @Override public void onError(Throwable e) { Log.i(TAG, "onError: "); } @Override public void onComplete() { Log.i(TAG, "onComplete: "); } };

订阅事件

observable.subscribe(observer);

到了这里就可以很成功的进行，完成一个完整的通路了

最后进行一站式直接操作

Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter<Integer> e) throws Exception e.onNext(1); e.onNext(2); e.onNext(3); e.onComplete(); } }).subscribe(new Observer<Integer>() { @Override public void onSubscribe(Disposable d) { Log.i(TAG, "onSubscribe: "); } @Override public void onNext(Integer integer) { Log.i(TAG, "onNext: "+integer); } @Override public void onError(Throwable e) { Log.i(TAG, "onError: "); } @Override public void onComplete() { Log.i(TAG, "onComplete: "); } });

只有subscribe()之后才能进行连通

顺便介绍下Consumer 如果你只是想处理接收到的数据可以使用Consumer 代码如下

Consumer<Integer> consumer = new Consumer<Integer>() { @Override public void accept(Integer integer) throws Exception { Log.i(TAG, "accept: " + integer); } };

订阅事件是一样的，observable不变

observable.subscribe(consumer);

2.rxjava2异步 rxjava2的异步处理真心非常棒 主要在于以下方法

.subscribeOn(Schedulers.newThread())//提供将Observable运行在一个新的线程中 .observeOn(AndroidSchedulers.mainThread())//提供将Observer运行在主线程中

很轻松的剥离非主线程执行的操作，很方便的对于网络，数据库，图片的处理 执行完整执行代码如下： observable，consumer还是上面的那个 像以下代码，如果切换两次线程，那么

observable.subscribeOn(Schedulers.newThread())//在这个线程执行 .subscribeOn(Schedulers.io())//不执行 .observeOn(AndroidSchedulers.mainThread())//执行 .observeOn(Schedulers.io())//切换到这个线程 .subscribe(consumer); //这里我们就可以灵活应用

rxjava2的内置线程，四种： —Schedulers.io() ：io操作的线程, 通常用于网络请求,读写文件 —Schedulers.computation()：需要大量计算的操作 —Schedulers.newThread() ：常规的新线程 —AndroidSchedulers.mainThread() ：Android的主线程