操作符总览

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进行定时计划

privateDisposablemDisposable;@OverrideprotectedvoiddoSomething(){mDisposable=Flowable.interval(20,TimeUnit.SECONDS).doOnNext(newConsumer<Long>(){@Overridepublicvoidaccept(@NonNullLongaLong)throwsException{doTask();}});}/***销毁时停止计划*/@OverrideprotectedvoidonDestroy(){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(newFunction<Response,Number>(){@OverridepublicMobileAddressapply(@NonNullResponseresponse)throwsException{if(response.isSuccessful()){ResponseBodybody=response.body();if(body!=null){Log.e(TAG,"map:转换前:"+response.body());returnnewGson().fromJson(body.string(),MobileAddress.class);}}returnnull;}}).observeOn(AndroidSchedulers.mainThread()).doOnNext(newConsumer<MobileAddress>(){@Overridepublicvoidaccept(@NonNullMobileAddresss)throwsException{Log.e(TAG,"doOnNext:Number："+s.getNumbser()+"\n");}})flatMap和concatMap

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

一对多事件转换：在flatMap集合中例如可以操作一个公司实体，并转换为单个部门实体，返回后在后续的accept中，又可以使用单个部门实体对每个成员进行逻辑处理。

实例：

Observable.fromArray(1,2,3,4,5).flatMap(newFunction<Integer,ObservableSource<Integer>>(){@OverridepublicObservableSource<Integer>apply(@NonNullIntegerinteger)throwsException{intdelay=0;if(integer==3){delay=500;//延迟500ms}returnObservable.just(integer*10).delay(delay,TimeUnit.MILLISECONDS);}}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(newConsumer<Integer>(){@Overridepublicvoidaccept(@NonNullIntegerinteger)throwsException{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,newBiFunction<Birth,Address,String>(){@OverridepublicStringapply(@NonNullBirthbirth,@NonNullAddressaddress)throwsException{return"合并后的数据为Birth："+birth.getResult()+"Address："+address.getResult();}}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(newConsumer<String>(){@Overridepublicvoidaccept(@NonNullStrings)throwsException{Log.e(TAG,"accept:成功："+s+"\n");}});过滤操作符操作符说明filter自定义筛选条件，返回booleandistinct去重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，都在返回falsetakeUntil当事件满足设定的条件时，该事件的下一个事件不会被发送了。包含超过临界条件的第一个事件takeWhile当事件满足设定的条件时，发送事件skipUntil直到设定的条件事件发出之后，开始发送原始事件。skipWhile跳过while范围内事件amb多个Observable序列中，只发送第一个contains是否存在特定元素exists是否满足特定条件DefaultIfEmpty如果没有正常结束事件（onComlete执行），返回默认值SequenceEqual判断两个事件序列是否是相同的数据，相同的顺序，相同的终止状态相似操作符对比

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