本篇文章已授权微信公众号 YYGeeker 独家发布转载请标明出处
CSDN学院课程地址
- RxJava2从入门到精通-初级篇:https://edu.csdn.net/course/detail/10036
- RxJava2从入门到精通-中级篇:https://edu.csdn.net/course/detail/10037
- RxJava2从入门到精通-进阶篇:https://edu.csdn.net/course/detail/10038
- RxJava2从入门到精通-源码分析篇:https://edu.csdn.net/course/detail/10138
4. RxJava基本响应类型
4.1 Observable
Observable是最基本的响应类型，但不支持背压，基本上适用大多数的应用场景
有关背压的概念等，都会在下一章介绍
public static void observable() {
//创建被观察者
Observable.create(new ObservableOnSubscribe<String>() {
@Override
//默认在主线程里执行该方法
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
e.onNext("俊俊俊很帅");
e.onNext("你值得拥有");
e.onNext("取消关注");
e.onNext("但还是要保持微笑");
e.onComplete();
}
})
//将被观察者切换到子线程
.subscribeOn(Schedulers.io())
//将观察者切换到主线程 需要在Android环境下运行
//.observeOn(AndroidSchedulers.mainThread())
//创建观察者并订阅
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
System.out.println("onNext=" + s);
}
@Override
public void onError(Throwable e) {
System.out.println("onError=" + e.getMessage());
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
输出
onNext=俊俊俊很帅
onNext=你值得拥有
onNext=取消关注
onNext=但还是要保持微笑
onComplete
4.2 Flowable
Flowable和Observable的使用基本相同，只不过Observable不支持背压，而Flowable支持背压。但需要注意的是，使用Flowable的时候，必须调用Subscription的requsest方法请求，不然上游是不会发射数据的
public static void flowable() {
//创建被观察者
Flowable.create(new FlowableOnSubscribe<String>() {
@Override
public void subscribe(FlowableEmitter<String> e) throws Exception {
e.onNext("俊俊俊很帅");
e.onNext("你值得拥有");
e.onNext("取消关注");
e.onNext("但还是要保持微笑");
e.onComplete();
}
}, BackpressureStrategy.DROP)
//将被观察者切换到子线程
.subscribeOn(Schedulers.io())
//将观察者切换到主线程 需要在Android环境下运行
//.observeOn(AndroidSchedulers.mainThread())
//创建观察者并订阅
.subscribe(new Subscriber<String>() {
@Override
public void onSubscribe(Subscription s) {
s.request(2);
}
@Override
public void onNext(String s) {
System.out.println("onNext=" + s);
}
@Override
public void onError(Throwable t) {
System.out.println("onError=" + t.getMessage());
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
输出
onNext=俊俊俊很帅
onNext=你值得拥有
4.3 Single
Single只发射一个元素，发射onSuccess或onError方法，所以没有complete方法，不像Observable或者Flowable，数据发射完成之后，需要调用complete告诉下游已经完成
public static void single() {
//创建被观察者
Single.create(new SingleOnSubscribe<String>() {
@Override
public void subscribe(SingleEmitter<String> e) throws Exception {
e.onSuccess("success");
}
})
//将被观察者切换到子线程
.subscribeOn(Schedulers.io())
//将观察者切换到主线程 需要在Android环境下运行
//.observeOn(AndroidSchedulers.mainThread())
//创建观察者并订阅
.subscribe(new SingleObserver<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onSuccess(String s) {
System.out.println("onSuccess=" + s);
}
@Override
public void onError(Throwable e) {
System.out.println("onError=" + e.getMessage());
}
});
}
输出
onSuccess=success
4.4 Completable
Completable不会发射数据，只会给下游发送一个信号。回调onComplete或onError方法
public static void completable() {
//创建被观察者
Completable.create(new CompletableOnSubscribe() {
@Override
public void subscribe(CompletableEmitter e) throws Exception {
e.onComplete();
}
})
//将被观察者切换到子线程
.subscribeOn(Schedulers.io())
//将观察者切换到主线程 需要在Android环境下运行
//.observeOn(AndroidSchedulers.mainThread())
//创建观察者并订阅
.subscribe(new CompletableObserver() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
@Override
public void onError(Throwable e) {
System.out.println("onError=" + e.getMessage());
}
});
}
输出
onComplete
4.5 Maybe
Maybe是Single和Completable的结合，需要注意的是onSuccess和onComplete方法只会执行其中一个，这不同于Observable和Flowable最后是以onComplete()结尾
public static void maybe() {
//创建被观察者
Maybe.create(new MaybeOnSubscribe<String>() {
@Override
public void subscribe(MaybeEmitter<String> e) throws Exception {
e.onSuccess("success");
e.onComplete();
}
})
//将被观察者切换到子线程
.subscribeOn(Schedulers.io())
//将观察者切换到主线程 需要在Android环境下运行
//.observeOn(AndroidSchedulers.mainThread())
//创建观察者并订阅
.subscribe(new MaybeObserver<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onSuccess(String s) {
System.out.println("onSuccess=" + s);
}
@Override
public void onError(Throwable e) {
System.out.println("onError=" + e.getMessage());
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
输出
onSuccess=success
4.6 小结
- Observable:最基本的响应类型，不支持背压
- Flowable:最基本的响应类型，支持背压
- Single:只发射一个数据的响应类型
- Completable:只发射一个信号的响应类型
- Maybe:Single和Completable的结合体
网友评论