理解RxJava里Observables的相关类型

Observable — Operator — Observer

Observable，可观察者，它就好像扬声器一样，做一些处理后发射出一些值。

Operator，操作符，它就像翻译员那样将数据从一种形式翻译/修改为另一种形式。

Observer，观察者，它获得那些由可观察者发射出来的或者有操作符转译的值。

它们的关系如图：

RxJava可观察类型

在RxJava的可观察类型如下：

Observable
Flowable
Single
Maybe
Completable

RxJava也提供了不同类型的观察者（Observer）对应于不同的可观察类型相：

Observer
SingleObserver
MaybeObserver
CompletableObserver

Observable <> Observer

最简单的Observable就是让它发射多个值。

示例：假设正在下载文件，它需要给出当前下载进度的百分比。这种情况我们就需要发射多个值。

创建一个简单的Observable

Observable<Integer> getDownloadObservable(final String url) {
    return Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) {
            if(!emitter.isDisposed()) {
                emitter.onNext(10);
            }
            if(!emitter.isDisposed()) {
                emitter.onNext(50);
            }
            if(!emitter.isDisposed()) {
                emitter.onNext(75);
            }
            if(!emitter.isDisposed()) {
                emitter.onNext(100);
                emitter.onCompleted();
            }
        }
    });
}

Observable对应的Observer

Observer<Integer> getObserver() {
    return new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
        }
        @Override
        public void onNext(Integer value) {
        }
        @Override
        public void onError(Throwable e) {
        }
        @Override
        public void onCompleted() {
        }
    };
}

Flowable <> Observer

当Observable发射了大量的值，而Observer处理不了那么多的值时，就可以用到Flowable。

这种情况下，Observable需要根据一些策略来跳过一些值，否则它会抛出异常。

可观察类型Flowable会使用策略来处理异常。

策略称为BackPressureStrategy（背压策略），异常为MissingBackPressureException（缺失背压异常）。

创建可观察的Flowable和Observer

和Observable类似，可以使用Flowable.create()来创建Flowable。与Flowable对应的Observer和上面的例子一样。

Single <> SingleObserver

Observable发射单值时使用Single。创建一个Single如下：

Single<String> getSingleObservable() {
    return Single.create(new SingleOnSubscribe<String>() {
        @Override
        public void subscribe(SingleEmitter emiiter) {
            if(!emitter.isDisposed()) {
                emitter.onSuccess("Hello");
            }
        }
    });
}

与Single对应的SingleObserver

SingleObserver<String> getSingleObserver() {
    return new SingleObserver<String>() {
        @Override
        public void onSubscribe(Disposable d) {
        }
        @Override
        public void onSuccess(String value) {
        }
        @Override
        public void onError(Throwable e) {
        }
    };
}

Maybe <> MaybeObserver

当Observable会发射一个值或者不会发射值，这时可以用到Maybe，创建Maybe如下：

Maybe<String> getMaybeObservable() {
    return Maybe.create(new MaybeOnSubscribe<String>() {
        @Override
        public void subscribe(MaybeEmitter emiiter) {
            if(!emitter.isDisposed()) {
                emitter.onSuccess("Hello");
            }
        }
    });
}

与Maybe对应的MaybeObserver

MaybeObserver<String> getMaybeObserver() {
    return new MaybeObserver<String>() {
        @Override
        public void onSubscribe(Disposable d) {
        }
        @Override
        public void onSuccess(String value) {
        }
        @Override
        public void onError(Throwable e) {
        }
        @Override
        public void onComplete() {
        }
    };
}

Completable <> CompletableObserver

当Observable需要处理一些任务当并没有发射任何值，这时可以用到Comletable。

创建Completable

Completable getCompletableObservable() {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(CompletableEmitter emiiter) {
            if(!emitter.isDisposed()) {
                emitter.onComplete();
            }
        }
    });
}

对应的CompletableObserver

CompletableObserver getCompletableObserver() {
    return new CompletableObserver() {
        @Override
        public void onSubscribe(Disposable d) {
        }
        @Override
        public void onError(Throwable e) {
        }
        @Override
        public void onCompleted() {
        }
    };
}

秒客网