理解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 getDownloadObservable(final String url) { return Observable.create(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter 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 getObserver() { return new Observer() { @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 getSingleObservable() { return Single.create(new SingleOnSubscribe() { @Override public void subscribe(SingleEmitter emiiter) { if(!emitter.isDisposed()) { emitter.onSuccess("Hello"); } } });}

与Single对应的SingleObserver

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

Maybe <> MaybeObserver

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

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

与Maybe对应的MaybeObserver

MaybeObserver getMaybeObserver() {

return new MaybeObserver() {

@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() { } };}