4章 RxJava基本响应类型

时间 2019-12-09

标签 rxjava 基本响应类型栏目 Java 繁體版

原文原文链接

本篇文章已受权微信公众号 YYGeeker 独家发布转载请标明出处bash

CSDN学院课程地址微信

RxJava2从入门到精通-初级篇:edu.csdn.net/course/deta…

RxJava2从入门到精通-中级篇:edu.csdn.net/course/deta…

RxJava2从入门到精通-进阶篇:edu.csdn.net/course/deta…

RxJava2从入门到精通-源码分析篇:edu.csdn.net/course/deta…

4. RxJava基本响应类型

4.1 Observable

Observable是最基本的响应类型，但不支持背压，基本上适用大多数的应用场景ide

有关背压的概念等，都会在下一章介绍源码分析

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");
                }
            });
}
复制代码

输出spa

onNext=俊俊俊很帅
onNext=你值得拥有
onNext=取消关注
onNext=但仍是要保持微笑
onComplete
复制代码

4.2 Flowable

Flowable和Observable的使用基本相同，只不过Observable不支持背压，而Flowable支持背压。但须要注意的是，使用Flowable的时候，必须调用Subscription的requsest方法请求，否则上游是不会发射数据的.net

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告诉下游已经完成code

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());
                }
            });
}
复制代码

输出server

onSuccess=success
复制代码

4.4 Completable

Completable不会发射数据，只会给下游发送一个信号。回调onComplete或onError方法ip

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的结合体