Rxjava操作符之功能性操作符
Rxjava2常见的功能性操作符有如下的一些:
废话不多说， 直接代码开路:
1.subscribe()
//功能性操作符
//subscribe()功能是连接被观察者Observable和观察者Observer 之间的桥梁
//创建被观察者对象Observable
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
//通过最基本的create操作符创建Observable被观察者对象
//并在subscribe方法中定义要发送的不同类型的事件
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onComplete();
}
});
//创建观察者对象Observer对象并重写对应的方法来处理
//被观察者发送的不同类型的事件行为
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
};
//通过subscribe()操作符来连接观察者Observer和被观察者Observable
observable.subscribe(observer);
//基于事件流的链式调用法
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("A");
e.onNext("B");
e.onNext("C");
e.onComplete();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: ");
}
@Override
public void onNext(String s) {
Log.d(TAG, "onNext: ");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: ");
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: ");
}
});2.线程切换操作符subscribeOn() / observerOn() 会单独开篇文章来进行讲解
3.延迟操作符 delay()
//延迟操作符delay()
//使得被观察者Observable延迟指定的时间再发送事件
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
Log.d(TAG, "subscribe thread = " + Thread.currentThread().getName());
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onNext(4);
}
//delay(t,TimeUnit,调度线程,错误延迟 如存在错误事件 true就是正常执行 执行后再抛出异常 false则反之
}).delay(2,TimeUnit.SECONDS,Schedulers.io(),true)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
});打印结果:
4: 在事件的生命周期中进行操作 --- do()操作符
//事件的整个生命周期操作符之 do()
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("A");
e.onNext("B");
e.onNext("C");
e.onNext("D");
// e.onError(new RuntimeException("发生了中断异常了!"));
e.onComplete();
}
}).doOnEach(new Consumer<Notification<String>>() {
@Override
public void accept(@NonNull Notification<String> stringNotification) throws Exception {
Log.d(TAG, "accept: 每次发送 next error complete 事件都会调用doOnEach : " + stringNotification.getValue());
}
}).doOnNext(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.d(TAG, "accept: 每次发送next事件前都会调用 doOnNext " + s);
}
}).doAfterNext(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.d(TAG, "accept: 每次发送next事件之后都会调用 doAfterNext " + s);
}
}).doOnComplete(new Action() {
@Override
public void run() throws Exception {
Log.d(TAG, "run: 每次发送complete事件之前都会调用 doOnComplete ");
}
}).doOnError(new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
Log.d(TAG, "accept: 每次发送error事件之前都会调用 doOnError" + throwable.getMessage());
}
}).doOnDispose(new Action() {
@Override
public void run() throws Exception {
Log.d(TAG, "run: 每次在中断观察者和被观察者连接之前都会调用doOnDispose");
}
}).doOnSubscribe(new Consumer<Disposable>() {
@Override
public void accept(@NonNull Disposable disposable) throws Exception {
Log.d(TAG, "accept: 每次在发生订阅前会调用 doOnSubscribe " + disposable.isDisposed());
}
}).doOnTerminate(new Action() {
@Override
public void run() throws Exception {
Log.d(TAG, "run: 每次在被观察者发送完所有事件或者即使出现错误中断了事件发送也同样会调用 doOnTerminate");
}
}).doOnLifecycle(new Consumer<Disposable>() {
@Override
public void accept(@NonNull Disposable disposable) throws Exception {
Log.d(TAG, "accept: doOnLifecycle ");
}
}, new Action() {
@Override
public void run() throws Exception {
Log.d(TAG, "run: doOnLifecycle");
}
}).doFinally(new Action() {
@Override
public void run() throws Exception {
Log.d(TAG, "run: 最后会调用 doFinally");
}
}).doAfterTerminate(new Action() {
@Override
public void run() throws Exception {
Log.d(TAG, "run: doAfterTerminate ");
}
}).subscribe(new Observer<String>() {
private Disposable mDisposable;
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
mDisposable = d;
}
@Override
public void onNext(String integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
// if(integer.equals("C")){
// if(!mDisposable.isDisposed()){
// mDisposable.dispose();
// }
// }
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: " + e.getMessage() + " thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
});5.错误处理操作符即在Observable发送事件的过程中遇到错误error时候的处理机制
5.1 onErrorReturn() / onErrorReturnItem()
//错误处理符之onErrorReturn() / onErrorReturnItem()
//onErrorReturnItem()捕捉被观察者发送的异常同时将异常转换成为next complete事件
//onErrorReturnItem()会返回一个固定预先设定好的值来代替捕捉到的错误
//而onErrorReturn()则提供Function让我们自定义处理捕捉错误转换的逻辑
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(11);
e.onNext(22);
e.onNext(33);
e.onError(new IndexOutOfBoundsException("严重的异常"));
}
})/*.onErrorReturnItem(100)*/ //针对所有的error异常事件都返回一个固定的值100 相当于将
//将我们的error事件转换成为了next事件和complete事件
//相当于调用了onNext() & onComplete()
//onErrorReturn是可以处理错误Error和异常Exception两种类型的异常的
//可以将这两种类型的事件都捕捉到并发送一个特殊的事件来进行处理
.onErrorReturn(new Function<Throwable, Integer>() {
@Override
public Integer apply(@NonNull Throwable throwable) throws Exception {
//通过Function的apply的方法捕捉我们的异常 并将异常
//转换成为next & complete事件 onNext(44) & onComplete()
Log.d(TAG, "apply: 捕捉到了异常 : " + throwable.getMessage());
return 44;
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: " + e.getMessage() + " thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
});打印结果:
5.2 onErrorResumeNext() / onExceptionResumeNext()
//错误处理符 onErrorResumeNext() / onExceptionResumeNext()
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("A");
e.onNext("B");
e.onNext("C");
e.onError(new Throwable("发生了一个错误!"));
// e.onError(new Exception("发生了一个异常"));
}
//onErrorResumeNext是可以处理Throwable 和 Exception两种类型的错误的
//onErrorResumeNext是可以将这两种类型的异常都转换成为新的Observable观察者对象并发送出去
}).onErrorResumeNext(new Function<Throwable, ObservableSource<? extends String>>() {
@Override
public ObservableSource<? extends String> apply(@NonNull Throwable throwable) throws Exception {
Log.d(TAG, "apply: 捕捉到了错误 e : " + throwable.getMessage() + " 将错误转换成了一个 新的Observable");
return Observable.just("D","E","F");
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
}
@Override
public void onNext(String integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: " + e.getMessage() + " thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
});
Log.d(TAG, "------------------------------------------------------------------------");
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(11);
e.onNext(22);
e.onNext(33);
e.onError(new Throwable("发送了一个错误!"));
// e.onError(new Exception("发生了一个异常！"));
}
//注意：onExceptionResumeNext拦截的是Exception类型的事件
//如果拦截的是Error类型的事件会将Error类型的事件传递给观察者Observer的onError方法
//并且不会将Error事件转换成一个Observable发送出去
//因此拦截转换Exception用onExceptionResumeNext
//拦截转换Error类型的事件用onErrorResumeNext
}).onExceptionResumeNext(new Observable<Integer>() {
@Override
protected void subscribeActual(Observer<? super Integer> observer) {
Log.d(TAG, "subscribeActual: 捕获到了异常将异常转换成为了一个新的Observable并发射出去!");
observer.onNext(44);
observer.onNext(55);
}
})/*.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())*/
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: " + e.getMessage() + " thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
});运行结果:
e.onError(new Throwable("发生了一个错误!"))
e.onError(new Exception("发生了一个异常"));
5.3 retry()重试功能操作符
//错误处理符之retry()
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onNext(4);
e.onNext(5);
e.onNext(6);
e.onError(new Throwable("发送了一个错误error"));
}
})//.retry()
// .retry(3)
/*.retry(new Predicate<Throwable>() {
@Override
public boolean test(@NonNull Throwable throwable) throws Exception {
Log.d(TAG, "test: retry重试 错误:" + throwable.getMessage());
if(!TextUtils.isEmpty(throwable.getMessage()) &&
TextUtils.equals(throwable.getMessage(),"发送了一个错误error")){
return true;
}else{
return false;
}
// return false; //返回false表示不要求Observable重新发送数据 而是直接调用观察者Observer的onError方法结束
// return true; //返回true表示要求Obsevable重新发送数据 若一直持续遇到错误就持续要求Observable重新发送数据
}
})*/
/*.retry(new BiPredicate<Integer, Throwable>() {
*//**
*
* @param integer 返回的重新要求Observable发送数据的次数
* @param throwable retry重试的错误类型
* @return
* @throws Exception
*//*
@Override
public boolean test(@NonNull Integer integer, @NonNull Throwable throwable) throws Exception {
Log.d(TAG, "test: retry 第" + integer + "次 重试错误:" + throwable.getMessage());
if(integer == 3){
return false;
}
return true;
}
})*/
//设置了一个总的重试次数,当超过了预先设置的重试次数的话
//不管Predicate中是返回true还是false都不会再去要求Observable在重新发送数据
.retry(3, new Predicate<Throwable>() {
@Override
public boolean test(@NonNull Throwable throwable) throws Exception {
return true;
}
})
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: " + e.getMessage() + " thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
});打印结果:
5.4 retryUntil()重试符
//重试操作符retryUntil()
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onError(new Exception("retryUntil错误！"));
}
}).retryUntil(new BooleanSupplier() {
@Override
public boolean getAsBoolean() throws Exception {
// return false; 返回false表示遇到错误会持续要求Observable重新发送数据
return true; //返回true表示遇到错误不会要求Observable重新发送数据而是直接调用Observer的onError方法
}
})/*.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())*/
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: " + e.getMessage() + " thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
});打印结果：
5.5 retryWhen()操作符
//错误重试操作符retryWhen()
//当捕捉到了上游发送的错误事件后会将错误事件转换成一个新的Observable被观察者对象
//并且通过新的被观察者对象发送的事件来决定是否需要重新订阅原被观察者对象发送的事件
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e)
throws Exception
{
e.onNext(1);
e.onNext(2);
e.onError(new Exception("发生错误了"));
e.onNext(3);
}
})
// 遇到error事件才会回调
.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Observable<Throwable> throwableObservable)
throws Exception
{
// 参数Observable<Throwable>中的泛型 = 上游操作符抛出的异常，可通过该条件来判断异常的类型
// 返回ObservableSource<?> = 新的被观察者 Observable（任意类型）
// 此处有两种情况：
// 1. 若 新的被观察者 Observable发送的事件 = Error事件，那么 原始Observable则不重新发送事件：
// 2. 若 新的被观察者 Observable发送的事件 = Next事件 ，那么原始的Observable则重新发送事件：
return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Throwable throwable)
throws Exception
{
// 1. 若返回的Observable发送的事件 = Error事件，则原始的Observable不重新发送事件
// 该异常错误信息可在观察者中的onError（）中获得
//return Observable.error(new Throwable("retryWhen终止啦"));
// 2. 若返回的Observable发送的事件 = Next事件，则原始的Observable重新发送事件（若持续遇到错误，则持续重试）
return Observable.just(1);
}
});
}
})
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG,
"onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread()
.getName());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG,
"onNext: " + integer + " thread = " + Thread.currentThread()
.getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG,
"onError: " + e.getMessage() + " thread = " + Thread.currentThread()
.getName());
}
@Override
public void onComplete() {
Log.d(TAG,
"onComplete: thread = " + Thread.currentThread().getName());
}
});新的被观察者Observable发送error事件终止订阅原被观察者对象发送的事件并且将新发送的error事件
回调给观察者Observer的onError方法:
新的被观察者Observable发送next事件重新订阅原被观察者对象发送的事件
(若持续收到上游发送的error事件则持续不断的订阅上游的Observable的事件):
6.重复操作符 repeat() / repeatWhen() / repeatuntil()
6.1 repeat()重复操作符
//重试操作符之repeat()
//repeat()操作符的特点是重复有限次数或者无限次数的去重新订阅被观察者Observable发送的事件
//但是前提是必须有complete事件来进行触发才会去 重新订阅和发送
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e)
throws Exception
{
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onComplete(); //complete事件才能够触发repeat操作符去重新订阅Obesrvable发送的事件
}
})
//repeat()无条件的重复无数次的接收被观察者Observable发送的事件
//但是要注意的是必须要接收到被观察者Observable发送的complete事件才能够
//重新触发订阅和发送
// .repeat()
// repeat(times) 设置重复订阅发送事件的次数,前提也必须是有complete事件来进行触发
.repeat(3)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: " + e.getMessage() + " thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
});打印结果:
6.2 repeatWhen()
//repeatWhen()操作符
//repeatWhen()操作符的特点是捕捉我们上游Observable对象发送的停止事件发送的标识(complete()/error())将其
//转换成一个新的Object类型的数据并传递给一个新的被观察者Observable对象以此来决定是否重新订阅发送原来上游的
//Observable对象
//两种情况:
//1.新的Observable发送complete()/error()不会重新订阅发送上游的Observable对象
//2.新的Observable发送除complete()/erro()事件以外的其他事件就会重新订阅发送上游的Observable对象
//注意:前提是上游的Observable对象发送了complete()事件来触发进行重新的订阅
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e)
throws Exception
{
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onNext(4);
//如果repeatWhen想要重复的订阅发送被观察者Observable必须有Observable的complete事件来进行触发
//否则即使条件满足了没有complete事件也是不能够重新进行订阅发送原Observable对象的
e.onComplete();
}
// 在Function函数中，必须对输入的 Observable<Object>进行处理，这里我们使用的是flatMap操作符接收上游的数据
}).repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
@Override
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 o)
throws Exception
{
// 情况1:若新被观察者Observable返回1个Complete()/Error()事件，则不重新订阅 & 发送原来的 Observable
// Observable.empty() = 发送Complete事件，但不会回调观察者Observer的onComplete()
// return Observable.empty();
// 返回Error事件 = 回调观察者Observer的onError()事件，并接收传过去的错误信息。
// return Observable.error(new Exception("发送error事件终止订阅!"));
// 情况2：若新被观察者（Observable）返回其余事件(比如Next事件)，则重新订阅 & 发送原来的 Observable
// 仅仅是作为1个触发重新订阅被观察者的通知,发送的是什么数据并不重要，只要不是Complete()/Error()事件即可
return Observable.just(1);
}
});
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: " + e.getMessage() + " thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
});打印结果:
新的被观察者发送complete() /error()事件
新的被观察者发送其它事件(非error/ complete):
6.3 repeatUntil()操作符
//repeatUntil()操作符
//repeatUntil()操作符是2.x新增的一个操作符，其表示的是直到满足某一个条件就不在重复发射数据
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onComplete();
// e.onError(new Throwable("repeatUntil异常..."));
}
}).repeatUntil(new BooleanSupplier() {
@Override
public boolean getAsBoolean() throws Exception {
//对于repeatUntil()操作符如果上游Observable没有发射complete事件是不会触发要求上游Observable重发数据的
//如果上游Observable发射了error事件也不会要求上游Observable重发数据(error事件是应该使用retryUntil())
return false; //返回false表示会重复发射上游Observable的数据
// return true; //返回true表示不会重复发射上游Observable的数据
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe: 开始采用subscribe连接 thread = " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer + " thread = " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError: " + e.getMessage() + " thread = " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: thread = " + Thread.currentThread().getName());
}
});打印结果:
关于RepeatWhen()和 RetryWhen()的区别和使用场景可以参考一下的两篇文章: