RxJava2 使用详解一之基础教程
目录
一、配置
二、原理
第一步：创建被观察者
第二步：创建观察者
第三步：建立订阅关系
链式操作
ObservableEmitter：
Disposable：
subscribe()重载方法
下篇：RxJava2 使用详解二之线程调度
一、配置
要在Android中使用RxJava2, 在app的build.gradle中添加依赖：
/*导入Rxjava RxAndroid 2018/8/30 最新版*/implementation 'io.reactivex.rxjava2:rxjava:2.2.1'implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'一个是Rxjava的库，一个是RxJava基于Android的库。
二、原理
RxJava原理详解
形象说明：用一条河的上游和下游代替被观察者和观察者, 上游和下游之间有个水闸，建立连接的情况下水闸是打开的，上游每发送一个事件，下游就能收到该事件。
RxJava的使用都是分三步
第一步：创建被观察者
//创建上游 Observable（被观察者） Observable observable = Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter emitter) throws Exception { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onComplete(); } }); 第二步：创建观察者
//创建下游 Observer（观察者） Observer observer = new Observer() { @Override public void onSubscribe(Disposable d) { Log.d(TAG, "onSubscribe"); } @Override public void onNext(Integer value) { Log.d(TAG, "onNext=" + value); } @Override public void onError(Throwable e) { Log.d(TAG, "onError"); } @Override public void onComplete() { Log.d(TAG, "onComplete"); } }; 第三步：建立订阅关系
//建立连接 observable.subscribe(observer);运行结果:
这里的事件发送的顺序是1,2,3, 事件接收的顺序也是1,2,3的顺序。
上游和下游就分别对应着RxJava中的Observable和Observer，它们之间的连接就对应着subscribe()。
只有当Observable(上游)和Observer(下游)建立连接之后, 上游才会开始发送事件. 也就是调用了subscribe() 方法之后才开始发送事件.
链式操作
把这段代码连起来写就成了被大家所熟知的RxJava的链式操作，运行结果也是一样的：
Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter emitter) throws Exception { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onComplete(); } }).subscribe(new Observer() { @Override public void onSubscribe(Disposable d) { Log.d(TAG, "onSubscribe"); } @Override public void onNext(Integer value) { Log.d(TAG, "onNext=" + value); } @Override public void onError(Throwable e) { Log.d(TAG, "onError"); } @Override public void onComplete() { Log.d(TAG, "onComplete"); } }); 解释一下两个关键词：ObservableEmitter和Disposable
ObservableEmitter：
Emitter就是发射器的，这个就是用来发出事件的，它可以发出三种类型的事件，通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、complete事件和error事件。
这里需要注意的是：
1、上游可以发送无限个onNext, 下游也可以接收无限个onNext。当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件。
2、当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件。上游可以不发送onComplete或onError。
3、发送多个onComplete是可以正常运行的, 但是收到第一个onComplete就不再接收了, 但若是发送多个onError, 则收到第二个onError事件会导致程序会崩溃。
我们就来一一试验一下：
1、发送多个onComplete
Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter emitter) throws Exception { Log.d(TAG, "emitter=1"); emitter.onNext(1); Log.d(TAG, "emitter=2"); emitter.onNext(2); Log.d(TAG, "emitter=3"); emitter.onNext(3); Log.d(TAG, "emitter=complete=1"); emitter.onComplete(); Log.d(TAG, "emitter=complete=2"); emitter.onComplete(); Log.d(TAG, "emitter=complete=3"); emitter.onComplete(); Log.d(TAG, "emitter=4"); emitter.onNext(4); } }).subscribe(new Observer() { @Override public void onSubscribe(Disposable d) { Log.d(TAG, "onSubscribe"); } @Override public void onNext(Integer value) { Log.d(TAG, "onNext: " + value); } @Override public void onError(Throwable e) { Log.d(TAG, "onError"); } @Override public void onComplete() { Log.d(TAG, "onComplete"); } }); 运行结果：
可以看到，发送多个onComplete是可以的，只是接收到一个onComplete之后就不在接收事件，但是上游依旧在发送事件。
2、发送多个onError事件
Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter emitter) throws Exception { Log.d(TAG, "emitter=1"); emitter.onNext(1); Log.d(TAG, "emitter=2"); emitter.onNext(2); Log.d(TAG, "emitter=3"); emitter.onNext(3);// Log.d(TAG, "emitter=complete=1");// emitter.onComplete();// Log.d(TAG, "emitter=complete=2");// emitter.onComplete();// Log.d(TAG, "emitter=complete=3");// emitter.onComplete(); Log.d(TAG, "emitter=error=1"); emitter.onError(new Throwable("no message")); Log.d(TAG, "emitter=error=2"); emitter.onError(new Throwable("no message")); Log.d(TAG, "emitter=error=3"); emitter.onError(new Throwable("no message")); Log.d(TAG, "emitter=4"); emitter.onNext(4); } }).subscribe(new Observer() { @Override public void onSubscribe(Disposable d) { Log.d(TAG, "onSubscribe"); } @Override public void onNext(Integer value) { Log.d(TAG, "onNext: " + value); } @Override public void onError(Throwable e) { Log.d(TAG, "onError="+e.getMessage()); } @Override public void onComplete() { Log.d(TAG, "onComplete"); } }); 运行结果：
从这个结果中可以看到，当收到第一个onError()后,打印了error信息但是程序并没有崩溃，然后第二个onError()程序就崩溃了。
Disposable：
这个单词的意思：是一次性的,用后就抛弃的. 那么在RxJava中怎么去理解它呢, 对应于上面的例子, 我们可以把它理解成水闸, 当调用它的dispose()方法时, 水闸关闭, 下游就收不到事件。
但是调用dispose()并不会让上游停止继续发送事件, 相反上游会继续发送剩余的事件.
上游依次发送1,2,3,complete,4,在下游收到第二个事件之后, 关闭水闸, 看看运行结果:
Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter emitter) throws Exception { Log.d(TAG, "emitter=1"); emitter.onNext(1); Log.d(TAG, "emitter=2"); emitter.onNext(2); Log.d(TAG, "emitter=3"); emitter.onNext(3); Log.d(TAG, "emitter=complete"); emitter.onComplete(); Log.d(TAG, "emitter=4"); emitter.onNext(4); } }).subscribe(new Observer() { private Disposable disposable; private int i; @Override public void onSubscribe(Disposable d) { Log.d(TAG, "onSubscribe"); disposable = d; } @Override public void onNext(Integer value) { Log.d(TAG, "onNext: " + value); i++; if (i == 2) { disposable.dispose(); Log.d(TAG, "isDisposed : " + disposable.isDisposed()); } } @Override public void onError(Throwable e) { Log.d(TAG, "onError"); } @Override public void onComplete() { Log.d(TAG, "onComplete"); } }); 运行结果为:
从运行结果我们看到, 在收到onNext(2)这个事件后, 关闭水闸, 但是上游仍然发送了onNext(3), complete, onNext(4)这几个事件, 而且上游并没有因为发送了onComplete而停止. 同时可以看到下游的onSubscribe()方法是最先调用的。
subscribe()重载方法
另外, subscribe()有多个重载的方法:
public final Disposable subscribe() {} public final Disposable subscribe(Consumer<? super T> onNext) {} public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {} public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {} public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {} public final void subscribe(Observer<? super T> observer) {}第一个不带任何参数的subscribe() 表示下游不关心上游发送的是什么。
第二个带有一个Consumer参数的方法表示下游只关心onNext事件, 其他的事件不关心, 因此我们如果只需要onNext事件可以这么写，代码如下:
Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter emitter) throws Exception { Log.d(TAG, "emitter=1"); emitter.onNext(1); Log.d(TAG, "emitter=2"); emitter.onNext(2); Log.d(TAG, "emitter=3"); emitter.onNext(3); Log.d(TAG, "emitter=complete"); emitter.onComplete(); Log.d(TAG, "emitter=4"); emitter.onNext(4); } }).subscribe(new Consumer() { @Override public void accept(Integer integer) throws Exception { Log.d(TAG, "accept: " + integer); } }); 运行结果如下：
接下来的几个方法的使用和第二个是一样的，就不多说了，最后一个方法的话，在文章前面就使用了，也不介绍了。
更多相关文章
- Android(安卓)ApiDemos示例解析(112):Views->Expandable Lists->
- Android截屏事件监听
- Android图集的上下左右拖动及动画效果研究，图片拖拽或视频拖拽
- wpa_supplicant适配层 -- 详解
- Android(安卓)JB 4.2 中InputManager 分发键盘消息给应用程序的
- Android磁盘管理-之vold源码分析(2)
- Hello Android(安卓)- PULL方式解析XML
- 【笔记】Android上ROS开发——android_core创建一个android应用
- touch事件处理