RxJava的Single、Completable以及Maybe

Rxjava2中Single,Completabley以及Maybe的内容；

参考地址: https://www.jianshu.com/p/45309538ad94

Single只发射单个数据或错误事件。

Maybe 能够发射0或者1个数据，要么成功，要么失败。有点类似于Optional

Completable 它从来不发射数据，只处理 onComplete 和 onError 事件。可以看成是Rx的Runnable。

Single的SingleObserver中只有onSuccess、onError，并没有onComplete。这是 Single 跟其他四种被观察者最大的区别。

Single.create(new SingleOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull SingleEmitter<String> e) throws Exception {

                e.onSuccess("test");
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                System.out.println(s);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(@NonNull Throwable throwable) throws Exception {
                throwable.printStackTrace();
            }
        });

上面的代码，由于Observer中有两个Consumer，还可以进一步简化成

Single.create(new SingleOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull SingleEmitter<String> e) throws Exception {

                e.onSuccess("test");
            }
        }).subscribe(new BiConsumer<String, Throwable>() {
            @Override
            public void accept(String s, Throwable throwable) throws Exception {
                System.out.println(s);
            }
        });

Single 可以通过toXXX方法转换成Observable、Flowable、Completable以及Maybe。

Completable 只有 onComplete 和 onError 事件，同时 Completable 并没有map、flatMap等操作符，它的操作符比起 Observable/Flowable 要少得多。

我们可以通过fromXXX操作符来创建一个Completable。这是一个Completable版本的Hello World。

Completable.fromAction(new Action() {
            @Override
            public void run() throws Exception {

                System.out.println("Hello World");
            }
        }).subscribe();

Completable 经常会结合andThen操作符

Completable.create(new CompletableOnSubscribe() {
            @Override
            public void subscribe(@NonNull CompletableEmitter emitter) throws Exception {

                try {
                    TimeUnit.SECONDS.sleep(1);
                    emitter.onComplete();
                } catch (InterruptedException e) {
                    emitter.onError(e);
                }
            }
        }).andThen(Observable.range(1, 10))
        .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(@NonNull Integer integer) throws Exception {
                System.out.println(integer);
            }
        });