delay():在被观察者发送事件前进行一些延迟的操作
// 1. 指定延迟时间
// 参数1 = 时间;参数2 = 时间单位
delay(long delay,TimeUnit unit)
// 2. 指定延迟时间 & 调度器
// 参数1 = 时间;参数2 = 时间单位;参数3 = 线程调度器
delay(long delay,TimeUnit unit,mScheduler scheduler)
// 3. 指定延迟时间 & 错误延迟
// 错误延迟,即:若存在Error事件,则如常执行,执行后再抛出错误异常
// 参数1 = 时间;参数2 = 时间单位;参数3 = 错误延迟参数
delay(long delay,TimeUnit unit,boolean delayError)
// 4. 指定延迟时间 & 调度器 & 错误延迟
// 参数1 = 时间;参数2 = 时间单位;参数3 = 线程调度器;参数4 = 错误延迟参数
delay(long delay,TimeUnit unit,mScheduler scheduler,boolean delayError): 指定延迟多长时间并添加调度器,错误通知可以设置是否延迟
do():在事件发送 & 接收的整个生命周期过程中进行操作 如发送事件前的初始化、发送事件后的回调请求等 如图总结
错误处理:发送事件过程中,遇到错误时的处理机制
onErrorReturn() 可捕获在它之前发生的异常
遇到错误时,发送1个新的Observable:
onErrorResumeNext()
onErrorResumeNext()
拦截的错误 =Throwable
;若需拦截Exception
请用onExceptionResumeNext()
- 若
onErrorResumeNext()
拦截的错误 =Exception
,则会将错误传递给观察者的onError
方法
onExceptionResumeNext()
onExceptionResumeNext()
拦截的错误 =Exception
;若需拦截Throwable
请用onErrorResumeNext()
如果onExceptionResumeNext()
拦截的错误 =Throwable
,则会将错误传递给观察者的onError
方法
retry()重试,即当出现错误时,让被观察者(Observable
)重新发射数据
<-- 1. retry() -->
// 作用:出现错误时,让被观察者重新发送数据
// 注:若一直错误,则一直重新发送
<-- 2. retry(long time) -->
// 作用:出现错误时,让被观察者重新发送数据(具备重试次数限制
// 参数 = 重试次数
<-- 3. retry(Predicate predicate) -->
// 作用:出现错误后,判断是否需要重新发送数据(若需要重新发送& 持续遇到错误,则持续重试)
// 参数 = 判断逻辑
<-- 4. retry(new BiPredicate<Integer, Throwable>) -->
// 作用:出现错误后,判断是否需要重新发送数据(若需要重新发送 & 持续遇到错误,则持续重试
// 参数 = 判断逻辑(传入当前重试次数 & 异常错误信息)
<-- 5. retry(long time,Predicate predicate) -->
// 作用:出现错误后,判断是否需要重新发送数据(具备重试次数限制
// 参数 = 设置重试次数 & 判断逻辑
retryUntil()出现错误后,判断是否需要重新发送数据
具体使用类似于retry(Predicate predicate)
,唯一区别:返回 true
则不重新发送数
retryWhen()遇到错误时,将发生的错误传递给一个新的被观察者(Observable
),并决定是否需要重新订阅原始被观察者(Observable
)& 发送事件
repeat()无条件地、重复发送 被观察者事件
repeat();不传入参数 = 重复发送次数 = 无限次
repeatWhen(Integer int )传入参数 = 重复发送次数有限
repeatWhen()有条件地、重复发送 被观察者事件
将原始 Observable
停止发送事件的标识(Complete()
/ Error()
)转换成1个 Object
类型数据传递给1个新被观察者(Observable
),以此决定是否重新订阅 & 发送原来的 Observable
- 若新被观察者(
Observable
)返回1个Complete
/Error
事件,则不重新订阅 & 发送原来的Observable
- 若新被观察者(
Observable
)返回其余事件时,则重新订阅 & 发送原来的Observable
// Observable.just(1,2,4).repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
@Override
// 在Function函数中,必须对输入的 Observable<Object>进行处理,这里我们使用的是flatMap操作符接收上游的数据
public ObservableSource<?> apply(@NonNull Observable<Object> objectObservable) throws Exception {
// 将原始 Observable 停止发送事件的标识(Complete() / Error())转换成1个 Object 类型数据传递给1个新被观察者(Observable)
// 以此决定是否重新订阅 & 发送原来的 Observable
// 此处有2种情况:
// 1. 若新被观察者(Observable)返回1个Complete() / Error()事件,则不重新订阅 & 发送原来的 Observable
// 2. 若新被观察者(Observable)返回其余事件,则重新订阅 & 发送原来的 Observable
return objectObservable.flatMap(new Function<Object, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Object throwable) throws Exception {
// 情况1:若新被观察者(Observable)返回1个Complete() / Error()事件,则不重新订阅 & 发送原来的 Observable
return Observable.empty();
// Observable.empty() = 发送Complete事件,但不会回调观察者的onComplete()
// return Observable.error(new Throwable("不再重新订阅事件"));
// 返回Error事件 = 回调onError()事件,并接收传过去的错误信息。
// 情况2:若新被观察者(Observable)返回其余事件,则重新订阅 & 发送原来的 Observable
// return Observable.just(1);
// 仅仅是作为1个触发重新订阅被观察者的通知,发送的是什么数据并不重要,只要不是Complete() / Error()事件
}
});
}
})
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "接收到了事件" + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应:" + e.toString());
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});