RxJava的简单使用
2018-11-21 本文已影响0人
kdong
导入RxJava的相关依赖
创建观察者与被观察者
链式调用
next、error、complete
subscribe的重载方法
指定观察者与被观察者所在的线程以及线程切换
结合Retrofit，实现一个简单的网络请求
一、引入依赖：
implementation "io.reactivex.rxjava2:rxjava:2.2.3"
implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'
二、观察者与被观察者
RxJava中的Observable和Observer通过SubScribe建立连接：
//创建一个上游 Observable：
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onError(new Throwable());
emitter.onComplete();
}
});
//创建一个下游 Observer
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "subscribe");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "" + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "error");
}
@Override
public void onComplete() {
Log.d(TAG, "complete");
}
};
//建立连接
observable.subscribe(observer);
运行结果：
11-21 16:49:58.443 21398-21398/com.example.qidong.testrxjava D/MainActivity: subscribe
11-21 16:49:58.443 21398-21398/com.example.qidong.testrxjava D/MainActivity: 1
11-21 16:49:58.444 21398-21398/com.example.qidong.testrxjava D/MainActivity: 2
11-21 16:49:58.444 21398-21398/com.example.qidong.testrxjava D/MainActivity: 3
11-21 16:49:58.444 21398-21398/com.example.qidong.testrxjava D/MainActivity: error
注意: 只有当上游和下游建立连接之后, 上游才会开始发送事件. 也就是调用了subscribe() 方法之后才开始发送事件.
三、RxJava的链式调用
使用RxJava的链式操作如下：
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(20);
emitter.onNext(30);
emitter.onError(new Throwable());
emitter.onNext(40);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG,"OnSubscribe");
}
@Override
public void onNext(Integer integer) {
Log.e(TAG,"onNext:"+integer);
}
@Override
public void onError(Throwable e) {
Log.e(TAG,"onError");
}
@Override
public void onComplete() {
Log.e(TAG,"OnComplete");
}
});
四、next、error、complete事件
从上面的代码中可以看到ObservableEmitter和Disposable。
ObservableEmitter：意思为发射器，可以通过调用onNext()、onError(Throwable)、onComplete()来发射next、error、complete这三种事件。
这其中需要注意的地方有：
上游可以发送无限个onNext, 下游也可以接收无限个onNext.
当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
上游可以不发送onComplete或onError.
最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然
Disposable, 这个单词的字面意思是一次性用品,用完即可丢弃的. 那么在RxJava中怎么去理解它呢, 对应于上面的水管的例子, 我们可以把它理解成两根管道之间的一个机关, 当调用它的dispose()方法时, 它就会将两根管道切断, 从而导致下游收不到事件.
注意: 调用dispose()并不会导致上游不再继续发送事件, 上游会继续发送剩余的事件.
来看个例子, 我们让上游依次发送1,2,3,complete,4,在下游收到第二个事件之后, 切断水管, 看看运行结果:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.e(TAG, "emit 1");
emitter.onNext(1);
Log.e(TAG, "emit 2");
emitter.onNext(2);
Log.e(TAG, "emit 3");
emitter.onNext(3);
Log.e(TAG, "onComplete");
emitter.onComplete();
Log.e(TAG, "emit 4");
emitter.onNext(4);
}
}).subscribe(new Observer<Integer>() {
private Disposable mDispoable;
private int i;
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "onSubscribe: ");
this.mDispoable=d;
}
@Override
public void onNext(Integer integer) {
Log.e(TAG, "onNext: "+integer);
i++;
if(i==2)
{
Log.e(TAG, "dispose");
mDispoable.dispose();
Log.e(TAG, "isDisposed:"+mDispoable.isDisposed());
}
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError");
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete");
}
});
运行结果：
11-21 17:11:50.338 23074-23074/com.example.qidong.testrxjava E/MainActivity: onSubscribe:
11-21 17:11:50.338 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 1
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: onNext: 1
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 2
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: onNext: 2
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: dispose
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: isDisposed:true
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 3
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: onComplete
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 4
可以看到，切断观察者和被观察者之间的关联之后，事件3、complete、4事件依然被发送了。
五、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) {}
最后一个带有Observer参数的我们已经使用过了,这里对其他几个方法进行说明.
不带任何参数的subscribe() 表示下游不关心任何事件,你上游尽管发你的数据去吧, 老子可不管你发什么.
带有一个Consumer参数的方法表示下游只关心onNext事件, 其他的事件我假装没看见, 因此我们如果只需要onNext事件可以这么写:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.e(TAG, "onNext:10");
emitter.onNext(10);
Log.e(TAG, "onNext:20");
emitter.onNext(20);
Log.e(TAG, "OnComplete");
emitter.onComplete();
Log.e(TAG, "onNext:30");
emitter.onNext(30);
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "accept:"+integer);
}
});
}
运行结果：
11-21 17:29:00.671 23664-23664/? E/MainActivity: onNext:10
11-21 17:29:00.672 23664-23664/? E/MainActivity: accept:10
11-21 17:29:00.672 23664-23664/? E/MainActivity: onNext:20
11-21 17:29:00.672 23664-23664/? E/MainActivity: accept:20
11-21 17:29:00.672 23664-23664/? E/MainActivity: OnComplete
11-21 17:29:00.672 23664-23664/? E/MainActivity: onNext:30
六、指定观察者与被观察者所在的线程以及线程切换
正常情况下，观察者和被观察者是工作在一个线程中的，也就是被观察者在哪个线程发射事件，观察者就在哪个线程接受事件：
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.e(TAG, "onNext:10");
emitter.onNext(10);
Log.e(TAG, "Observable Thread Name:"+Thread.currentThread().getName());
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "accept:"+integer);
Log.e(TAG, "Observer Thread Name:"+Thread.currentThread().getName());
}
});
运行结果：
11-21 17:43:25.366 25090-25090/? E/MainActivity: onNext:10
11-21 17:43:25.366 25090-25090/? E/MainActivity: accept:10
11-21 17:43:25.366 25090-25090/? E/MainActivity: Observer Thread Name:main
11-21 17:43:25.366 25090-25090/? E/MainActivity: Observable Thread Name:main
一般而言，我们会将耗时操作放在子线程中去做，然后在主线程中根据运行的结果去更新UI。
RxJava中提供了线程调度器实现这样的需求：我们在子线程中发射事件，在主线程中接受事件:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.e(TAG, "onNext:10");
emitter.onNext(10);
Log.e(TAG, "Observable Thread Name:"+Thread.currentThread().getName());
}
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "accept:"+integer);
Log.e(TAG, "Observer Thread Name:"+Thread.currentThread().getName());
}
});
此时再看一下运行结果：
11-21 17:50:37.408 25766-25934/com.example.qidong.testrxjava E/MainActivity: onNext:10
11-21 17:50:37.409 25766-25934/com.example.qidong.testrxjava E/MainActivity: Observable Thread Name:RxNewThreadScheduler-1
11-21 17:50:37.453 25766-25766/com.example.qidong.testrxjava E/MainActivity: accept:10
11-21 17:50:37.453 25766-25766/com.example.qidong.testrxjava E/MainActivity: Observer Thread Name:main
简单的来说, subscribeOn() 指定的是上游发送事件的线程, observeOn() 指定的是下游接收事件的线程.
多次指定被观察者的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.
多次指定观察者的线程是可以的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次.
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.e(TAG, "onNext:10");
emitter.onNext(10);
Log.e(TAG, "Observable Thread Name:"+Thread.currentThread().getName());
}
}).subscribeOn(Schedulers.newThread())
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.observeOn(Schedulers.io())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "accept:"+integer);
Log.e(TAG, "Observer Thread Name:"+Thread.currentThread().getName());
}
});
再来看下运行结果：
11-21 17:54:36.940 26399-26522/com.example.qidong.testrxjava E/MainActivity: onNext:10
11-21 17:54:36.941 26399-26522/com.example.qidong.testrxjava E/MainActivity: Observable Thread Name:RxNewThreadScheduler-1
11-21 17:54:36.973 26399-26524/com.example.qidong.testrxjava E/MainActivity: accept:10
11-21 17:54:36.974 26399-26524/com.example.qidong.testrxjava E/MainActivity: Observer Thread Name:RxCachedThreadScheduler-2
可以看出观察者所在的线程被修改成功。
为了看出观察者所在线程的切换过程，我们可以添加Log去查看：
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.e(TAG, "emit 10");
emitter.onNext(10);
Log.e(TAG, "emit 20");
emitter.onNext(20);
}
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "After Change to MainThread,thread name:"+Thread.currentThread().getName());
}
}).observeOn(Schedulers.io()).doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "After Change to IO,thread name:"+Thread.currentThread().getName());
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e(TAG, "accept: "+integer);
}
});
运行结果：
11-21 18:08:37.913 28956-29018/com.example.qidong.testrxjava E/MainActivity: emit 10
11-21 18:08:37.914 28956-29018/com.example.qidong.testrxjava E/MainActivity: emit 20
11-21 18:08:37.945 28956-28956/com.example.qidong.testrxjava E/MainActivity: After Change to MainThread,thread name:main
11-21 18:08:37.946 28956-28956/com.example.qidong.testrxjava E/MainActivity: After Change to MainThread,thread name:main
11-21 18:08:37.946 28956-29020/com.example.qidong.testrxjava E/MainActivity: After Change to IO,thread name:RxCachedThreadScheduler-1
11-21 18:08:37.946 28956-29020/com.example.qidong.testrxjava E/MainActivity: accept: 10
11-21 18:08:37.947 28956-29020/com.example.qidong.testrxjava E/MainActivity: After Change to IO,thread name:RxCachedThreadScheduler-1
11-21 18:08:37.947 28956-29020/com.example.qidong.testrxjava E/MainActivity: accept: 20
在RxJava中, 已经内置了很多线程选项供我们选择, 例如有
Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
Schedulers.newThread() 代表一个常规的新线程
AndroidSchedulers.mainThread() 代表Android的主线程
这些内置的Scheduler已经足够满足我们开发的需求, 因此我们应该使用内置的这些选项, 在RxJava内部使用的是线程池来维护这些线程, 所有效率也比较高.
