RxJava2.x 萌新之路 操作符篇

操作符总览

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)

实例：取代handler进行定时计划

private Disposable mDisposable;
    @Override
    protected void doSomething() {
        mDisposable = Flowable.interval(20, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        doTask();
                    }
                });
    }

    /**
     * 销毁时停止计划
     */
    @Override
    protected void onDestroy() {
        super.onDestroy();
        if (mDisposable != null){
            mDisposable.dispose();
        }
    }

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, Number>() {
           @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

两者都可以实现数据集合中一对多事件的转换，后者会按发送的顺序获取接收结果，前者可能是乱序接收（不确定哪个事件先完成）。

一对多事件转换：在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");
                    }
                });

过滤操作符

其他操作符

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秒后）

条件操作符

相似操作符对比

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