Rxjava总结

最近看了Rxjava的相关知识点，现在总结一下以备之后查看。

Rxjava比较重要的就是操作符：

http://blog.csdn.net/maplejaw_/article/details/52396175

创建操作的操作符有：

Create   Just   From   Defer 

Empty/Never/Throw

Interval   Range   Repeat   Start   Timer

create： 使用OnSubscribe从头创建一个Observable，这种方法比较简单。需要注意的是，使用该方法创建时，建议在OnSubscribe#call方法中检查订阅状态，以便及时停止发射数据或者运算。

from： 将一个Iterable, 一个Future, 或者一个数组，内部通过代理的方式转换成一个Observable。Future转换为OnSubscribe是通过OnSubscribeToObservableFuture进行的，Iterable转换通过OnSubscribeFromIterable进行。数组通过OnSubscribeFromArray转换。

just： 将一个或多个对象转换成发射这个或这些对象的一个Observable。如果是单个对象，内部创建的是ScalarSynchronousObservable对象。如果是多个对象，则是调用了from方法创建。

empty： 创建一个什么都不做直接通知完成的Observable

error： 创建一个什么都不做直接通知错误的Observable

never： 创建一个什么都不做的Observable

defer： 只有当订阅者订阅才创建Observable，为每个订阅创建一个新的Observable。内部通过OnSubscribeDefer在订阅时调用Func0创建Observable。

interval： 创建一个按照给定的时间间隔发射从0开始的整数序列的Observable，内部通过OnSubscribeTimerPeriodically工作。

Observable.interval(1, TimeUnit.SECONDS).subscribe(newAction1() {

  @Override

  public void call(Long aLong) {//每隔1秒发送数据项，从0开始计数//0,1,2,3....}

});

range： 创建一个发射指定范围的整数序列的Observable

timer： 创建一个在给定的延时之后发射数据项为0的Observable,内部通过OnSubscribeTimerOnce工作

转换操作的操作符有：

Map   FlatMap   GroupBy  Buffer  Scan   Window 

过滤操作的操作符有：

Debounce   Distinct   ElementAt   Filter   Take   Takelast   

First   IgnoreElements   Last   Sample   Skip   SkipLast

filter： 过滤数据。内部通过OnSubscribeFilter过滤数据。

Observable.just(3,4,5,6) .filter(newFunc1() {

@Override

public Boolean call(Integer integer) {

    return integer>4;               

 }           

 }) .subscribe(item->Log.d("JG",item.toString()));//5,6

ofType： 过滤指定类型的数据，与filter类似，

Observable.just(1,2,"3").ofType(Integer.class) .subscribe(item -> Log.d("JG",item.toString()));

take： 只发射开始的N项数据或者一定时间内的数据。内部通过OperatorTake和OperatorTakeTimed过滤数据。

Observable.just(3,4,5,6)            .take(3)//发射前三个数据项.take(100, TimeUnit.MILLISECONDS)//发射100ms内的数据

takeLast： 只发射最后的N项数据或者一定时间内的数据。内部通过OperatorTakeLast和OperatorTakeLastTimed过滤数据。takeLastBuffer和takeLast类似，不同点在于takeLastBuffer会收集成List后发射。

Observable.just(3,4,5,6) .takeLast(3) .subscribe(integer -> Log.d("JG",integer.toString()));//4,5,6

takeFirst：提取满足条件的第一项。内部实现源码如下：

publicfinalObservabletakeFirst(Func1 predicate) {returnfilter(predicate).take(1);//先过滤，后提取}

first/firstOrDefault：只发射第一项（或者满足某个条件的第一项）数据，可以指定默认值。

Observable.just(3,4,5,6) .first() .subscribe(integer -> Log.d("JG",integer.toString()));//3Observable.just(3,4,5,6)              .first(newFunc1() {@OverridepublicBooleancall(Integer integer) {returninteger>3;                  }              }) .subscribe(integer -> Log.d("JG",integer.toString()));//4

last/lastOrDefault：只发射最后一项（或者满足某个条件的最后一项）数据，可以指定默认值。

skip：跳过开始的N项数据或者一定时间内的数据。内部通过OperatorSkip和OperatorSkipTimed实现过滤。

skipLast：跳过最后的N项数据或者一定时间内的数据。内部通过OperatorSkipLast和OperatorSkipLastTimed实现过滤。

elementAt/elementAtOrDefault：发射某一项数据，如果超过了范围可以的指定默认值。内部通过OperatorElementAt过滤。

Observable.just(3,4,5,6)                .elementAt(2)        .subscribe(item->Log.d("JG",item.toString()));//5

ignoreElements：丢弃所有数据，只发射错误或正常终止的通知。内部通过OperatorIgnoreElements实现。

distinct：过滤重复数据，内部通过OperatorDistinct实现。

Observable.just(3,4,5,6,3,3,4,9)      .distinct()      .subscribe(item->Log.d("JG",item.toString()));//3,4,5,6,9

distinctUntilChanged：过滤掉连续重复的数据。内部通过OperatorDistinctUntilChanged实现

Observable.just(3,4,5,6,3,3,4,9)      .distinctUntilChanged()      .subscribe(item->Log.d("JG",item.toString()));//3,4,5,6,3,4,9

throttleFirst：定期发射Observable发射的第一项数据。内部通过OperatorThrottleFirst实现。

throttleWithTimeout/debounce：发射数据时，如果两次数据的发射间隔小于指定时间，就会丢弃前一次的数据,直到指定时间内都没有新数据发射时

sample/throttleLast：定期发射Observable最近的数据。内部通过OperatorSampleWithTime实现。

timeout： 如果原始Observable过了指定的一段时长没有发射任何数据，就发射一个异常或者使用备用的Observable。

组合操作的操作符有：

Zip   Merge   StartWith   CombinLastest   Join   SwitchOnNext   

concat： 按顺序连接多个Observables。需要注意的是Observable.concat(a,b)等价于a.concatWith(b)。

startWith： 在数据序列的开头增加一项数据。startWith的内部也是调用了concat

merge： 将多个Observable合并为一个。不同于concat，merge不是按照添加顺序连接，而是按照时间线来连接。其中mergeDelayError将异常延迟到其它没有错误的Observable发送完毕后才发射。而merge则是一遇到异常将停止发射数据，发送onError通知。

zip： 使用一个函数组合多个Observable发射的数据集合，然后再发射这个结果。如果多个Observable发射的数据量不一样，则以最少的Observable为标准进行压合。内部通过OperatorZip进行压合。

zip工作流程

combineLatest： 。当两个Observables中的任何一个发射了一个数据时，通过一个指定的函数组合每个Observable发射的最新数据（一共两个数据），然后发射这个函数的结果。类似于zip，但是，不同的是zip只有在每个Observable都发射了数据才工作，而combineLatest任何一个发射了数据都可以工作，每次与另一个Observable最近的数据压合。具体请看下面流程图。

combineLatest工作流程

错误处理的操作符有：

Catch   Retry