RxJava操作符介绍，部分使用，使用场景，源码浅析（二）
2016-08-11 本文已影响305人
梦sora
1. RxJava操作符介绍
创建操作
- Create 从头创建一个Observable
- Defer 直到有观察者订阅时才创建Observable，并且为每个观察者创建一个新的Observable
- Empty/Never/Throw 创建行为受限的特殊Observable
Empty 创建一个不发射任何数据但是正常终止的Observable
Never 创建一个不发射数据也不终止的Observable
Throw 创建一个不发射数据以一个错误终止的Observable - From 将其它对象或数据结构创建为Observable挨个发射
- Interval 创建一个固定间隔发射整数序列的Observable，替代Timer
- Just 创建一个发射指定值的Observable
- Range 创建发射指定范围的整数序列的Observable
- Repeat 创建一个发射特定数据重复多次的Observable
- Timer 创建在一个指定的延迟之后发射单个数据的Observable
变换操作（对Observable发射的数据进行变换）
- Buffer 缓存，定期收集Observable的数据放进一个数据包裹，然后发射这些数据包裹，而不是一次发射一个值
- FlatMap 将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable
- GroupBy 将原来的Observable分拆为Observable集合，将原始Observable发射的数据按Key分组，每一个Observable发射一组不同的数据
- Map 通过对序列的每一项都应用一个函数变换Observable发射的数据，实质是对序列中的每一项执行一个函数，函数的参数就是这个数据项
- Scan 对Observable发射的每一项数据应用一个函数，然后按顺序依次发射这些值
- Window 定期将来自Observable的数据分拆成一些Observable窗口，然后发射这些窗口，而不是每次发射一项
过滤操作
- Debounce 仅在过了一段指定的时间还没发射数据时才发射一个数据
- Distinct 过滤掉重复数据项
- ElementAt 取特定位置的数据项
- Filter 过滤掉函数返回false的数据项
- First 只发射满足条件的第一条数据
- IgnoreElements 忽略所有的数据，只保留终止通知(onError或onCompleted)
- Last 只发射最后一条数据
- Sample 定期发射最新的数据，等于是数据抽样，有的实现里叫ThrottleFirst
- Skip 跳过前面的若干项数据
- SkipLast 跳过后面的若干项数据
- Take 只保留前面的若干项数据
- TakeLast 只保留后面的若干项数据
组合操作
- CombineLatest 当两个Observables中的任何一个发射了数据时，使用一个函数结合每个Observable发射的最近数据项，并且基于这个函数的结果发射数据
- Join 无论何时，如果一个Observable发射了一个数据项，只要在另一个Observable发射的数据项定义的时间窗口内，就将两个Observable发射的数据合并发射
- Merge 将两个Observable发射的数据组合并成一个
- StartWith 在发射原来的Observable的数据序列之前，先发射一个指定的数据序列或数据项
- Switch 将一个发射多个Observables的Observable转换成另一个单独的Observable，如果Observable正在发射数据的时候，源Observable又发射出一个新的Observable，则前一个Observable发射的数据会被抛弃，直接发射新的Observable所发射的数据
- Zip 通过一个函数将多个Observables的发射物结合到一起，基于这个函数的结果为每个结合体发射单个数据项。Zip操作符将多个Observable发射的数据按顺序组合起来，每个数据只能组合一次，而且都是有序的。最终组合的数据的数量由发射数据最少的Observable来决定。
错误处理
- Catch 继续序列操作，将错误替换为正常的数据，从onError通知中恢复
- Retry 如果Observable发射了一个错误通知，重新订阅它，期待它正常终止。在发生错误的时候会重新进行订阅,而且可以重复多次，所以发射的数据可能会产生重复。如果重复指定次数还有错误的话就会将错误返回给观察者
RetryWhen 指示Observable遇到错误时，将错误传递给另一个Observable来决定是否要重新给订阅这个Observable，新Observable处理错误，老的继续流程。
辅助操作
- Delay 延迟一段时间发射结果数据
- Do 注册一个动作占用一些Observable的生命周期事件，相当于Mock某个操作，Do操作符就是给Observable的生命周期的各个阶段加上一系列的回调监听
- Materialize/Dematerialize 将发射的数据和通知都当做数据发射，或者反过来
- ObserveOn 指定观察者观察Observable的调度程序（工作线程）
- Serialize 强制Observable按次序发射数据并且功能是有效的
- Subscribe 收到Observable发射的数据和通知后执行的操作
- SubscribeOn 指定Observable应该在哪个调度程序上执行
- TimeInterval 将一个Observable转换为发射两个数据之间所耗费时间的Observable
- Timeout 添加超时机制，如果过了指定的一段时间没有发射数据，就发射一个错误通知
- Timestamp 给Observable发射的每个数据项添加一个时间戳
- Using 创建一个只在Observable的生命周期内存在的一次性资源
我们创建一个资源并使用它，用一个Observable来限制这个资源的使用时间，当这个Observable终止的时候，这个资源就会被销毁。
条件和布尔操作
- All 判断Observable发射的所有的数据项是否都满足某个条件
- Amb 给定多个Observable，只让第一个发射数据的Observable发射，将至多9个Observable结合起来，让他们竞争。哪个Observable首先发射了数据（包括onError和onComplete)就会继续发射这个Observable的数据，其他的Observable所发射的数据都会被丢弃。
- Contains 判断Observable是否会发射一个指定的数据项，如果源Observable已经结束了却还没有发射这个数据则返回false
- IsEmpty操作符用来判断源Observable是否发射过数据，如果发射过就会返回false，如果源Observable已经结束了却还没有发射这个数据则返回true。
- DefaultIfEmpty 发射来自原始Observable的数据，如果原始Observable没有发射数据，就发射一个默认数据
- SequenceEqual 判断两个Observable是否按相同的数据序列
- SkipUntil 丢弃原始Observable发射的数据，直到标志的Observable发射了一个数据，然后发射原始Observable的剩余数据
- SkipWhile 丢弃原始Observable发射的数据，直到一个特定的条件为假，然后发射原始Observable剩余的数据
- TakeUntil 发射来自原始Observable的数据，直到标志的Observable发射了一个数据，然后跳过剩余的数据
- TakeWhile 发射原始Observable的数据，直到一个特定的条件为真，然后跳过剩余的数据
算术和聚合操作
- Average 计算Observable发射的数据序列的平均值，然后发射这个结果
- Concat 不交错的连接多个Observable的数据
将多个Observable结合成一个Observable并发射数据，并且严格按照先后顺序发射数据，前一个Observable的数据没有发射完，是不能发射后面Observable的数据的。 - Count 计算Observable发射的数据个数，然后发射这个结果
如果源Observable发射错误，则会将错误直接报出来；在源Observable没有终止前，count是不会发射统计数据的。 - Max 计算并发射数据序列的最大值
- Min计算并发射数据序列的最小值
- Reduce 按顺序对数据序列的每一个应用某个函数，然后返回这个值
接收Observable发射的数据和函数的计算结果作为下次计算的参数，输出最后的结果。 - Sum 计算并发射数据序列的和
连接操作
- Connect 指示一个可连接的Observable开始发射数据给订阅者
- Publish 将一个普通的Observable转换为可连接的
- RefCount 使一个可连接的Observable表现得像一个普通的Observable
- Replay 确保所有的观察者收到同样的数据序列，即使他们在Observable开始发射数据之后才订阅
自定义操作符
- lift 如果我们的自定义操作符想要作用到Observable发射出来的数据上，使用lift操作符
- compose 如果我们的自定义操作符想要改变整个的Observable，使用compose操作符
2. RxJava操作符的使用
3. lift源码解析
1.先看看Observable的create()
public static <T> Observable<T> create(OnSubscribe<T> f) {
return new Observable<T>(RxJavaHooks.onCreate(f));
}
Observable的构造方法
protected Observable(OnSubscribe<T> f) {
this.onSubscribe = f;
}
和RxJavaHooks的onCreate()方法
public static <T> Observable.OnSubscribe<T> onCreate(Observable.OnSubscribe<T> onSubscribe) {
Func1<OnSubscribe, OnSubscribe> f = onObservableCreate;
if (f != null) {
return f.call(onSubscribe);
}
return onSubscribe;
}
hook了Observable.OnSubscribe
2.在看Observable的lift()
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
return create(new OnSubscribeLift<T, R>(onSubscribe, operator));
}
OnSubscribeLift类
public final class OnSubscribeLift<T, R> implements OnSubscribe<R> {
final OnSubscribe<T> parent;
final Operator<? extends R, ? super T> operator;
public OnSubscribeLift(OnSubscribe<T> parent, Operator<? extends R, ? super T> operator) {
this.parent = parent;
this.operator = operator;
}
@Override
public void call(Subscriber<? super R> o) {
try {
Subscriber<? super T> st = RxJavaHooks.onObservableLift(operator).call(o);
try {
st.onStart();
parent.call(st);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
st.onError(e);
}
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
o.onError(e);
}
}
}
RxJavaHooks的onObservableLift()方法
public static <T, R> Operator<R, T> onObservableLift(Operator<R, T> operator) {
Func1<Operator, Operator> f = onObservableLift;
if (f != null) {
return f.call(operator);
}
return operator;
}
同样hook了Observable.OnSubscribe，是新的onSubscribe
这里的parent就是旧onSubscribe
总结下，新的Observable会像一个代理一样，负责接收原始的Observable发出的事件，并在处理后发送给Subscriber
4. RxJava应用场景
- 取数据先检查缓存的场景
- 界面需要等到多个接口并发取完数据，再更新merge
- 一个接口的请求依赖另一个API请求返回的数据
- 界面按钮需要防止连续点击的情况throttleFirst
- 响应式的界面checkedChanges
- 复杂的数据变换
- 轮询
- 线程切换
- RxBus
- RxBinding Android控件对RxJava的支持库
- rx-preferences 使SharedPreferences支持 RxJava
- RxLifecycle 帮助RxJava在Android中生命周期的控制，避免内存溢出等问题
- retrofit Retrofit
- storio 数据库对RxJava的支持
