前言
之前的几天写了 1,2,3,4,5,6 篇关于 RxJava 的操作符的基本用法的演示，琢磨琢磨了一下，感觉不太全，再补充点功能类的操作符，也就是常用的，但是不太在意的操作符。
正文
1、subscribe 操作符
1)、作用
订阅，连接观察者 & 被观察者，组成订阅关系
2)、代码
/**
* subscribe 操作符
*/
@SuppressLint("CheckResult")
private void subscribeMethod() {
Observable.just(123, 234, 455, 677)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
logDUtils("accept:" + integer);
}
});
}这里就不贴 效果了，有了 subscribe 操作符，才可以让 被观察者与观察者联系到一起。
2、线程调度类 subscribeOn 和 observeOn
1)、作用
subscribeOn ：被观察者 的运行线程
observeOn ： 观察者 的运行线程
两个操作符 内的参数：
Schedulers.immediate ( ) ： 默认线程--------------> 当前线程 不指定是什么具体线程；
AndroidSchedulers.mainThread ( ) ：主线程
Schedulers.newThread ( ) ：常规新工作线程 非主线程
Schedulers.io ( ) ：i o 操作符线程 -----------------> 多用于网络请求、读写文件等 io 密集型操作
Schedulers.computation( ) ： cpu 计算线程
建议查看笔者仰慕的大佬 Season_zlc 的 给初学者的RxJava2.0教程(二) 这篇文章
2)、代码
/**
* 线程调度类 操作符以及参数
*/
@SuppressLint("CheckResult")
private void threadMethod() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
logDUtils("create thread 1：" + Thread.currentThread().getName());
emitter.onNext("a");
logDUtils("create thread 2：" + Thread.currentThread().getName());
emitter.onNext("b");
}
//先将被观察者 设置运行与 工作线程
}).subscribeOn(Schedulers.io())
//将 被观察者 运行在 主线程
.subscribeOn(AndroidSchedulers.mainThread())
//将 观察者 首先运行在 工作线程
.observeOn(Schedulers.newThread())
//将 观察者转换到 主线程
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
logDUtils("accept thread：" + Thread.currentThread().getName() + " 打印数据： " + s);
}
});
}3)、效果
注意：
从效果内 可以看出 被观察者 对象 只能 设置 1 次，而 观察者所在线程 可以进行多次改变。
3、delay 操作符
1)、作用
将被观察者延迟一段时间再发送事件
2)、代码演示
/**
* delay 操作符
*/
private void delayMethod() {
Observable.just(1, 2, 3, 4, 5).delay(2, TimeUnit.SECONDS)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
logDUtils("onSubscribe");
}
@Override
public void onNext(Integer integer) {
logDUtils("onNext: " + integer);
}
@Override
public void onError(Throwable e) {
logDUtils("onError" + e.getMessage());
}
@Override
public void onComplete() {
logDUtils("onComplete");
}
});
}3)、效果
其他 重载方法
// 参数:时间 时间单位
public final Observable<T> delay(long delay, TimeUnit unit)
// 参数: 时间 时间单位 线程调度器
public final Observable<T> delay(long delay, TimeUnit unit, Scheduler scheduler)
// 参数: 时间 时间单位 错误延迟参数
public final Observable<T> delay(long delay, TimeUnit unit, boolean delayError)
// 参数:时间 时间单位 线程调度器 错误延迟参数
public final Observable<T> delay(long delay, TimeUnit unit, Scheduler scheduler, boolean delayError)4、do 类 操作符
doOnEach：当被观察者每发送 1 次数据事件就会调用 1 次；
doOnNext：执行 onNext 事件前调用；
doAfterNext： 执行onNext事件后调用；
doOnComplete：被观察者正常发送事件完毕后调用；
doOnError： 被观察者发送错误事件时调用；
doOnSubscribe： 观察者订阅时调用；
doAfterTerminate：被观察者发送事件完 毕后调用，不管是正常终止还是非正常终止；
doFinally：最后执行
doOnUnsubscribe(): 取消订阅执行
5、onErrorReturn 操作符
1)、作用
当发生错误时， 发送一个约定好的事件， 保证正常停止。
2)、代码
/**
* onErrorReturn 操作符
*/
private void onErrorReturnMethod() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("a");
emitter.onError(new Throwable());
emitter.onNext("b");
}
}).onErrorReturn(new Function<Throwable, String>() {
@Override
public String apply(Throwable throwable) throws Exception {
logDUtils("发送错误");
return "c";
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
logDUtils("onSubscribe");
}
@Override
public void onNext(String s) {
logDUtils("onNext:" + s);
}
@Override
public void onError(Throwable e) {
logDUtils("onError:" + e.getMessage());
}
@Override
public void onComplete() {
logDUtils("onComplete:");
}
});
}3)、效果
当发送 onError 事件之后，后面的 b 并没有被发送，而是直接走 onErrorReturn 操作符发送 c ，结束程序。
6、onErrorResumeNext 操作符
作用
当发生错误的时候 发送 一个被观察者。 记住 与上面时有区别的哦！
7、onExceptionResumeNext 操作符
作用
当发生 异常 的时候 发送一个被观察者。记住这个与上面也不一样哈！
8、 retry 操作符
作用
当发生 错误 或者 异常时，重新让被观察者发送 一 次数据
重载方法 有 5个
public final Observable<T> retry()
// 参数 = 重试次数
public final Observable<T> retry(long times)
// 参数 = 判断逻辑
public final Observable<T> retry(Predicate<? super Throwable> predicate)
// 参数 = 判断逻辑（传入当前重试次数 & 异常错误信息）
public final Observable<T> retry(BiPredicate<? super Integer, ? super Throwable> predicate)
// 参数 = 设置重试次数 & 判断逻辑
public final Observable<T> retry(long times, Predicate<? super Throwable> predicate)9、retryUntil 操作符
作用
出现错误后，判断是否需要重新发送数据
若需要，则持续进行重试操作 类似于 public final Observable<T> retry(Predicate<? super Throwable> predicate) 方法
10、retryWhen 操作符
作用
遇到错误时，将发生的错误传递给一个新的被观察者，并决定是否需要重新订阅 原 被观察者 并且发送 原 事件；
11、repeat 操作符
作用
无参 ：表示 无条件重复 发送被观察 的事件
有参：表示 重复次数
12、repeatWhen 操作符
有条件的 进行 重复发送
好了，马上双十一， 筒子们，准备好购物车，一起来剁手吧，额，还是再添两本书吧。
