RxJava2.x萌新之路操作符篇

操作符总览

创新互联专注于企业成都全网营销推广、网站重做改版、德令哈网站定制设计、自适应品牌网站建设、H5网站设计、电子商务商城网站建设、集团公司官网建设、外贸营销网站建设、高端网站制作、响应式网页设计等建站业务，价格优惠性价比高，为德令哈等各大城市提供网站开发制作服务。

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() {
                    @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() {
           @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() {
             @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>() {
                    @Override
                    public ObservableSource 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() {
            @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() {
            @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() {
                    @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)：出现错误时，根据指定逻辑(可以捕获重发的次数和发生的错误)决定是否让被观察者重新发送数据
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()：用于事件流中，可以延迟发送事件流中的某一次发送。

网站栏目：RxJava2.x萌新之路操作符篇
本文路径：http://scjbc.cn/article/igejjh.html