优先参考：ReactiveX - Operators ReactiveX/RxJava文档中文版-操作符
ReactiveX官网 - Operators介绍部分翻译
RxJava 3.x系列（三）RxJava操作符使用场景举例
- 使用RxJava时从问题场景出发选定合适的操作符，首先判断需要怎样的变换，参数变换 or Observable变换 or 错误处理...
- 根据分类从中文翻译网、Reactive官网找出合适的操作符，或者自定义操作符，从而满足自身需求.
一、ReactiveX核心操作符分类：11类
-
Creating 创建操作 -
Create/Defer/From/Just/Start/Repeat/Range -
Transforming 变换操作 -
Buffer/Window/Map/FlatMap/GroupBy/Scan -
Filtering 过滤操作 -
Debounce/Distinct/Filter/Sample/Skip/Take -
Combining 结合操作 -
And/StartWith/Join/Merge/Switch/Zip -
Error Handling 错误处理 -
Catch/Retry -
Utility 辅助操作 -
Delay/Do/ObserveOn/SubscribeOn/Subscribe -
Conditional and Boolean 条件与布尔操作 -
All/Amb/Contains/SkipUntil/TakeUntil -
Mathematical and Aggregate 算术与聚合操作 -
Average/Concat/Count/Max/Min/Sum/Reduce -
Backpressure Flowable
-
Connectable 连接操作 -
Connect/Publish/RefCount/Replay -
Convert 转换操作 -
ToFuture/ToList/ToMap/ToIterable/toMultiMap
二、操作符链式调用
以Observable.create().map().subscribeOn().observeOn().subscribe()调用链为例，所有操作符的调用都将生成新的Observable对象，并将原始调用操作符的Observable对象作为成员变量传入，形成Observable一层包一层效果。如：Observable.create()创建ObservableCreate对象，ObservableCreate.map()创建ObservableMap对象并传入ObservableCreate对象作为其成员变量。ObservableMap.subscribeOn()创建ObservableSubscribeOn对象并传入ObservableMap对象，observeOn()调用后也同理，最终生成的Observable对象为ObservableObserveOn。
当subscribe(observer)被调用后，触发ObservableObserveOn类的subscribeActual(observer)方法，subscribeActual()在上述所有Observable对象中都有独立实现。ObservableObserveOn中已传入上一级Observable对象即ObservableSubscribeOn，subscribeActual()中将上级Observable对象订阅ObserveOnObserver，代码为source.subscribe(new ObserveOnObserver<>(observer, w, delayError, bufferSize))，这里ObserveOnObserver包含下游Observer。ObserveOnObserver为ObservableObserveOn静态内部类，每个操作符生成的新Observable类中都定义了自己的Observer内部类来封装下游observer，即通过调用subscribeActual(observer)传入下游observer给自身，将下游observer封装为自己的observer内部类对象，供上游Observable对象订阅。继而上游逐级订阅上游直到最顶层。
订阅到最上层为ObservableCreate.subscribe(MapObserver)，随即触发ObservableCreate的subscribeActual(MapObserver)，将执行ObservableOnSubscribe.subscribe(CreateEmitter(observer))，CreateEmitter.onNext()开始发射数据，observer.onNext()逐级回调，直到最下层。
总结：上游每执行一个操作符都将生成一个新的Observable对象，执行下一个操作符时会将上一个操作符生成的Observable对象作为成员变量传入新的Observable对象，当subscribe()方法调用后，Observer将被逐级封装为新的Observer，进而被逐级订阅，形成链式订阅。新的Observer中封装操作符的各种变换操作，当最上层Observable发射onNext()，将形成链式调用或者叫做链式回调
注：Observable一层包一层是为了形成Observer逐级订阅效果，Observer一层包一层是为了在上下游onNext()执行之间封装操作符逻辑
Observable抽象类的subscribe()方法如下。所有Observable对象都将调用此方法订阅下游Observer，并调用subscribeActual()
public final void subscribe(@NonNull Observer<? super T> observer) {
Objects.requireNonNull(observer, "observer is null");
try {
...
subscribeActual(observer);
} catch (NullPointerException e) {
...
}
ObservableObserveOn类的subscribeActual()方法，所有Observable类都实现了自身subscribeActual()方法，用于在链式调用中订阅上下游
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<>(observer, w, delayError, bufferSize));
}
}
调用链Observable.create().map().subscribeOn().observeOn().subscribe()如下图所示：
observable_observer.jpg
最终形成的Observable订阅关系: ObservableCreate.subscribe(MapObserver)
最上游开始发射数据 & 链式调用开始触发: ObservableOnSubscribe.subscribe(CreateEmitter(MapObserver))
开始发射数据: emitter.onNext(T t)
observer.onNext()逐级调用如下所示：
CreateEmitter(MapObserver).onNext() -> MapObserver(SubscriveOnObserver).onNext() -> SubscribeObserver(ObserveOnObserver).onNext() -> ... Observer().onNext()
操作符内部实现的各种操作将通过插入到Observer中onNext()的逐级调用之间，从而形成丰富多彩的链式变换操作
注意：Observable操作符的顺序很重要。其它模式如Builder模式也能形成链式调用，但Builder模式的调用顺序通常不重要，如下引用自ReactiveX官网
三、操作符介绍
1.变换操作
（1）map: 将Observable发射的每一项数据做一个变换
map.png
MapObserver onNext()方法：
public void onNext(T t) {
...
U v;
...
try {
// 将原有数据变换为新数据
v = Objects.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
downstream.onNext(v); // 传递新参数继续onNext发射数据
}
（2）flatMap：将上游Observable变换为新的Observable
flatMap.png
.flatMap(new Function<String, ObservableSource<Integer>>() {...})
MergeObserver onNext()方法：
public void onNext(T t) {
// safeguard against misbehaving sources
...
ObservableSource<? extends U> p;
try {
// mapper.apply(t)返回flatMap中实现的ObservableSource
p = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null ObservableSource");
...
if (maxConcurrency != Integer.MAX_VALUE) {
synchronized (this) {
if (wip == maxConcurrency) {
sources.offer(p);
return;
}
wip++;
}
}
subscribeInner(p); // 将MergeObserver封装成InnerObserver给ObservableSource订阅
}
2.过滤操作
（1）filter: 只发射通过指定条件测试的数据项
filter.png
（2）skip: 丢弃Observable发射的前N项数据(参数传入数字) or 丢弃开始发射数据的那段时间发射的数据(参数传入时间间隔)
skip.png
3.组合操作
(1) zip: 将多个Observable发送的事件结合到一起，下游可收到上游两个Observable发射的数据。
注：它只发射与发射数据项最少的那个Observable一样多的数据。
zip.png
ObservableZip的subscribeActual()方法
public void subscribeActual(Observer<? super R> observer) {
ObservableSource<? extends T>[] sources = this.sources;
int count = 0;
if (sources == null) {
sources = new ObservableSource[8];
for (ObservableSource<? extends T> p : sourcesIterable) {
if (count == sources.length) {
ObservableSource<? extends T>[] b = new ObservableSource[count + (count >> 2)];
System.arraycopy(sources, 0, b, 0, count);
sources = b;
}
sources[count++] = p;
}
} else {
count = sources.length;
}
if (count == 0) {
EmptyDisposable.complete(observer);
return;
}
ZipCoordinator<T, R> zc = new ZipCoordinator<>(observer, zipper, count, delayError); // 传入下游observer
zc.subscribe(sources, bufferSize); // 开始订阅两个上游Observable
}
ZipCoordinator subscribe()方法
public void subscribe(ObservableSource<? extends T>[] sources, int bufferSize) {
ZipObserver<T, R>[] s = observers;
int len = s.length;
for (int i = 0; i < len; i++) {
s[i] = new ZipObserver<>(this, bufferSize); // 供给上游两个Observable订阅
}
// this makes sure the contents of the observers array is visible
this.lazySet(0);
downstream.onSubscribe(this);
for (int i = 0; i < len; i++) {
if (cancelled) {
return;
}
sources[i].subscribe(s[i]); // sources[i]即为两个Observable，s[i]为两个ZipObserver，用于缓存和处理两个上游Observable的数据发射
}
}
（2）merge: 将多个Observables的输出合并，合并后内部使用flatMap生成新的Observable
merge.png
4.错误处理
（1）onErrorReturn: 让Observable遇到错误时发射一个特殊的项并且正常终止。此操作符返回一个镜像原有Observable行为的新Observable，后者会忽略前者的onError调用，不会将错误传递给观察者，作为替代，它会发射一个特殊数据项并调用观察者的onCompleted方法。
onErrorReturn.png
（2）retry: 发生错误导致onError()将发生，随即再给一次机会无错误地完成它的数据序列。retry总是传递onNext()通知给观察者
retry.png
5.条件与布尔操作
（1）all: 判定是否Observable发射的所有数据都满足某个条件，上游dispose或者调用onComplete下游才收到回调
all.png
（2）contains: 判定一个Observable是否发射一个特定的值
contains.png
6.算术与聚合操作
（1）count: 将一个Observable转换成一个发射单个值的Observable，这个值表示原始Observable发射数据的数量
count.png
（2）reduce: 按顺序对Observable发射的每项数据应用一个函数并发射最终的值
reduce.png
7.连接操作
（1）publish: 将普通的Observable转换为可连接的Observable (connectable Observable)，它并不会在被订阅时开始发射数据，而是直到使用了connect操作符时才会开始。用这种方法，你可以在任何时候让一个Observable开始发射数据。
原理：publish调用后订阅链将被切断，connect调用后将切断的订阅链重新订阅完整
publish_connect.png
（2）connect: 让一个可连接的Observable开始发射数据给订阅者，如上图
ObservablePublish的connect()方法
public void connect(Consumer<? super Disposable> connection) {
boolean doConnect = false;
PublishConnection<T> conn;
for (;;) {
conn = current.get();
if (conn == null || conn.isDisposed()) {
PublishConnection<T> fresh = new PublishConnection<>(current);
if (!current.compareAndSet(conn, fresh)) {
continue;
}
conn = fresh;
}
doConnect = !conn.connect.get() && conn.connect.compareAndSet(false, true);
break;
}
try {
connection.accept(conn);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
throw ExceptionHelper.wrapOrThrow(ex);
}
if (doConnect) {
source.subscribe(conn); // publish操作符切断订阅链，劫持真实subscribe到connect操作
}
}
8.转换操作 类似于变换操作
1.toList：将Observable发射的多项数据组合成一个List，然后调用一次onNext方法传递整个列表
toList.png
2.toFuture：将Observable转换为一个返回单个数据项的Future
toFuture.png
四、BackPressure
物理学概念：运动流体在密闭容器中沿其路径流动时，由于管道变细或受到障碍物、急转弯道的阻碍而被施加的与运动方向相反的力。
异步订阅关系中存在被观察者发送事件速度与观察者接收事件速度不匹配的情况，故引出下游控制上游流速的手段：下游请求多少个数据上游就发射多少个数据
大数据流用Flowable，小数据流用Observable。Flowable默认事件数量缓存大小为128，上游发射数据到缓存队列，队列数据如果下游来不及处理，将抛出MissingBackpressureException异常
Flowable第一行代码：
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) {
Log.d(TAG, "emit 1");
emitter.onNext(1);
Log.d(TAG, "emit 2");
emitter.onNext(2);
Log.d(TAG, "emit 3");
emitter.onNext(3);
Log.d(TAG, "emit complete");
emitter.onComplete();
}
}, BackpressureStrategy.ERROR) // 策略参数
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "onSubscribe");
mSubscription = s;
s.request(1); // 给上游发通知请求上游发射数据
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer);
mSubscription.request(1); // 给上游发通知请求上游发射数据
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
相比于Observable，Flowable替换Dispose为Subscription，并增加BackPressure策略参数，如上BackpressureStrategy.ERROR即默认策略；BackpressureStrategy.BUFFER策略提供更大的事件缓存队列空间；BackpressureStrategy.DROP策略在队列满后丢弃新来的事件，新事件不进入队列；BackpressureStrategy.LATEST策略在队列满后保留新来的事件，将队列头部事件丢弃，故Latest策略总是能获取到最后最新的事件。不同策略将生成不同的Emmiter，如下FlowableCreate的subscribeActual()方法所示：
public void subscribeActual(Subscriber<? super T> t) {
BaseEmitter<T> emitter;
switch (backpressure) {
case MISSING: {
emitter = new MissingEmitter<>(t);
break;
}
case ERROR: {
emitter = new ErrorAsyncEmitter<>(t);
break;
}
case DROP: {
emitter = new DropAsyncEmitter<>(t);
break;
}
case LATEST: {
emitter = new LatestAsyncEmitter<>(t);
break;
}
default: {
emitter = new BufferAsyncEmitter<>(t, bufferSize());
break;
}
}
t.onSubscribe(emitter);
try {
source.subscribe(emitter);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
emitter.onError(ex);
}
}
DropAsyncEmitter onNext()方法代码
public final void onNext(T t) {
if (isCancelled()) {
return;
}
if (t == null) {
onError(ExceptionHelper.createNullPointerException("onNext called with a null value."));
return;
}
if (get() != 0) { // 队列容量为0则抛出异常
downstream.onNext(t);
BackpressureHelper.produced(this, 1);
} else {
// nothing to do
}
}
LatestAsyncEmitter onNext()方法代码
public void onNext(T t) {
if (done || isCancelled()) {
return;
}
if (t == null) {
onError(ExceptionHelper.createNullPointerException("onNext called with a null value."));
return;
}
queue.set(t);
drain();
}
P.S. 本文源码基于RxJava 3.0.4，除首图外其余图片引用自ReactiveX官网
By the way: 非常感谢能提出批评和修改意见
contact: tu.wentai@nexuslink.cn
网友评论