Rxjava2-二、操作符

Rxjava记录总结操作符:创建操作符、转换操作符、合并操作符、过滤操作符、其他操作符、条件操作符.

创建操作符

just
自动一次发送事件序列

Observable .just("1", "2", "3", "4", "5", "6", "7", "8", "9", "10") 
依次发送调用onNext（），最后默认调用complete()

create
手动创建事件序列,返回一个可自由操作的emitter,有点是自用控制事件流程.

emitter.onNext(); 
emitter.onError(); 
emitter.onComplete();

fromIterable
传入数组并按角标一次发送事件

Observable.fromIterable(list)，每次接收单个元素。

fromArray
传入数组一次性发送，一次接收所有元素。

timer
延时发送事件

Observable .timer(2, TimeUnit.SECONDS)

interval
可取代CountDownTimer、Handler，5秒发送一次事件： 倒计时操作

Observable .interval(5, TimeUnit.SECONDS)

intervalRange
给事件更多的时间控制：
intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit)
参数1：起始发送值
参数2：发送数量
参数3：首次发送延迟事件
参数4：每次发送事件间隔
参数5：时间单位

Range
依次发送范围内的事件

Observable.range(2, 6)，接收类型Integer

转换操作符

map
实现单个数据的转换
实例:把网络中ResponseBody用Gson转换为相对应的数据实体再下发给子类。

.map(new Function<Response, MobileAddress.class>() {
           @Override
           public MobileAddress apply(@NonNull Response response) throws Exception {
            if (response.isSuccessful()) {
                ResponseBody body = response.body();
                if (body != null) {
                    Log.e(TAG, "map:转换前:" + response.body());
                    return new Gson().fromJson(body.string(), MobileAddress.class);
                }
            }
                    return null;
                }
            }).observeOn(AndroidSchedulers.mainThread())
            .doOnNext(new Consumer<MobileAddress>() {
             @Override
             public void accept(@NonNull MobileAddress s) throws Exception {
                 Log.e(TAG, "doOnNext: Number：" + s.getNumbser() + "\n");
             }
         })

flatMap和concatMap
两个都可以实现数据集合中一对多事件的转换,concatMap会按发送的顺序获取接收结果,flatMap可能是乱序接收(不确定哪个事件先完成)
一对多事假转换:在flatMap集合中例如可以操作一个公司实体,并转换为单个部门实体,返回在后续的accept中,又可以使用单个部门实体对每个成员进行逻辑处理.
实例:

Observable.fromArray(1,2,3,4,5)
                .flatMap(new Function<Integer, ObservableSource<Integer>>() {
                    @Override
                    public ObservableSource<Integer> apply(@NonNull Integer integer) throws Exception {

                        int delay = 0;
                        if(integer == 3){
                            delay = 500;//延迟500ms
                        }
                        return Observable.just(integer *10).delay(delay, TimeUnit.MILLISECONDS);
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(@NonNull Integer integer) throws Exception {
                Log.e("tag","accept:"+integer);
            }
        });

使用flatMap结果：10,20,40,30,50
使用contactMap结果：10,20,30,40,50

buffer
分批发送事件

Observable .just(1, 2, 3, 4, 5, 6) .buffer(2) 
发送1,2；发送3,4；在发送5,6

合并操作符

merge和contat
两者都可以合并多个Observable事件，前者发送顺序不确定（并行无序），后者按顺序发送（串行有序）。
mergeArray和concatArray效果相同，适用于大于4个事件的情况。
实例：
定义cache和network两个事件，先查看缓存是否有数据，有即onNext去刷新页面，没有则onComplete读取网络数据。

 Observable.concat(cache,network)

concatDelayError和 mergeDelayError
两者都可以在merge和contat操作中出现错误时停止发送当前事件集合，但不影响合并中的另一个事件集合发送

zip

zip 操作符可以将多个 Observable 的数据结合为一个数据源再发射出去
实例：分别请求生日、地址、性别等信息后，将多个请求结果合成一个，再进行UI更新。

....分别请求生日、地址...
Observable.zip(observable1, observable2, new BiFunction<Birth, Address, String>() {
            @Override
            public String apply(@NonNull Birth birth, @NonNull Address address) throws Exception {
                return "合并后的数据为 Birth："+birth.getResult()+" Address："+address.getResult();
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {
                        Log.e(TAG, "accept: 成功：" + s+"\n");
                    }
                });

过滤操作符

filter: 自定义筛选条件，返回boolean
distinct : 去重
distinctUntilChanged: 过滤连续相同事件
skip，skipLast:跳过前n个事件或最后n个
take和takeLast:只接收前n个事件或最后n个
elementAt和elementAtOrError:前者只发送第n个，可设置默认值，不抛异常；后者越界抛异常。
ignoreElements：只接收完成和报错信息
distinct:去重
ofType:指定接收数据类型
throttleFirst/throttleLast:只接收指定时间内第一个或最后一个事件

其他操作符

do

doOnEach() :当Observable每发送一次事件就会调用一次(包含onNext()，onError()，onComplete())
doOnNext(): 执行 onNext()前调用
doAfterNext()： 执行onNext()后调用
doOnComplete()：执行onComplete()前调用
doOnError():执行 onError()前调用
doOnTerminate(): 执行终止(无论正常发送完毕/异常终止)
doFinally(): 最后执行
doOnSubscribe()：观察者订阅是调用
doOnUnScbscribe()： 观察者取消订阅时调用

onErrorReturn

捕获错误返回,不发送后续事件

onExceptionResumeNext/onErrorResumeNext

捕获错误跳过当前事件同时不中断发送后续事件。

retry

retry() ： 出现错误时，让被观察者重新发送数据。若错误一直发生，则一直重新发送
retry(long time)：与retry不同的书，若错误一直发生，被观察者则一直重新发送数据，但这持续重新发送有次数限制
retry(Predicate predicate) ： 出现错误时，根据指定逻辑(可以捕获到发生的错误)决定是否让被观察者重新发送数据
retry(new BiPredicate<Integer, Throwable>)：出现错误时，根据指定逻辑(可以捕获重发的次数和发生的错误)决定是否让被观察者重新发送数据
retry（long time,Predicate predicate） ： 出现错误时，根据指定逻辑(可以捕获到发生的错误)决定是否让被观察者重新发送数据。并且有持续重发的次数限制

retryUntil

遇到错误时根据制定规则选择是否重发

retryWhen

遇到错误时，将发生的错误传递给一个新的被观察者(Observable)，并决定是否需要重新订阅原始被观察者(Observable)

repeat和repeatWhen

repeat重复发射 observable的数据序列，可以使无限次也可以是指定次数.不传时为重复无限次。
repeatWhen遇到错误选择返回object给新观察者或中止事件
返回参数选择：
Observable.empty()；
发送Complete事件，但不会回调观察者的Complete（）
onComplete（）
直接完成。
Observable.error(new Throwable("不再重新订阅事件"));
Observable.just(1);
继续发送事件。

debounce

一定的时间内没有操作就会发送事件（只会发送最后一次操作的事件）
实例：

Observable.intervalRange(1, 2, 3, 4, TimeUnit.SECONDS) 
.debounce(2, TimeUnit.SECONDS) 
只有最后一个4的事件会被发送（2秒后）

条件操作符

all:判断被观察者所有事件是否满足某个事件，如果全部满足则返回true，都在返回false
takeUntil:
当事件满足设定的条件时，该事件的下一个事件不会被发送了。包含超过临界条件的第一个事件
takeWhile:
当事件满足设定的条件时，发送事件
skipUntil
直到设定的条件事件发出之后，开始发送原始事件。
skipWhile
跳过while范围内事件
amb
多个Observable序列中，只发送第一个
contains
是否存在特定元素
exists
是否满足特定条件
DefaultIfEmpty
如果没有正常结束事件（onComlete执行），返回默认值
SequenceEqual
判断两个事件序列是否是相同的数据，相同的顺序，相同的终止状态

相似操作符对比

timer()：用于创建Observable，延迟发送一次。
interval()：用于创建Observable，跟TimerTask类似，用于周期性发送。
delay()：用于事件流中，可以延迟发送事件流中的某一次发送。