RxJava操作符续

Posted by 程序亦非猿 on 2015-10-16

前言

RxJava处理数据主要还是靠不同的操作符,在这里对RxJava比较常用的操作符做必要的介绍,附加一些自己的理解.

操作符有难有易,部分操作符非常简单容易理解,只记录些许文字
部分操作符复杂,可能会配有弹珠图,它十分有助于对操作符的理解,而理解弹珠图,前提是知道每个元素的意义,特给出下图:

一定要理解

OK,准备就绪,开启RxJava操作符之旅吧!

创建

这里指的是Observable的创建

create

使用一个函数从头创建一个Observable

从头创建意味着onNext onCompleted onError都需要我们自己调用.

定制度最高,也最麻烦,需要时刻判断observer.isUnsubscribed()来判断是否继续处理发射数据.

defer

只有当订阅者订阅才创建Observable;为每个订阅创建一个新的Observable

当被订阅时,为每个订阅创建一个新的Observable,所以defer可以确保Observable包含最新的数据。

from

将一个Iterable, 一个Future, 或者一个数组转换成一个Observable

from操作符可以转换Future、Iterable和数组。对于Iterable和数组,产生的Observable会发射Iterable或数组的每一项数据

相对于create,defer,from生成Observable非常方便,加上iterable和数组在Java中的使用率,使得from出场的频率非常高!

just

将一个或多个对象转换成发射这个或这些对象的一个Observable,很方便

range

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

  • range(int start,int count)
  • range(int,int,Scheduler)

range(3,10)表示从3开始,发射10个数字,即3,4,5….12

interval

创建一个按照给定的时间间隔发射整数序列的Observable

  • interval(long,TimeUnit)
  • interval(long,TimeUnit,Scheduler)

默认在computation调度器上

timer

创建一个在给定的延时之后发射单个数据的Observable

empty/never/throw

  • empty 创建一个什么都不做直接通知完成的Observable
  • throw 创建一个什么都不做直接通知错误的Observable
  • never 创建一个不发射任何数据的Observable

    暂时没有用到过

小结

就创建Observable而言,from,just,defer,create用得多一些,一般来讲掌握这几个即可.

变换操作

变换是RxJava处理数据最强大的部分,一定要理解,要活用

buffer

  • buffer(int count)
  • buffer(int count,int skip)

buffer是将数据缓存成一个List,再发送

  • count 集合数量的的大小
  • skip 每次发射后扔掉数据的个数(看后面例子好理解一些)

举例:

1
Observable.just(1,2,3,4,5,6,7,8,9,10).buffer(3).subscribe(subscriber);

其实buffer(count)调用的是buffer(count,count)!,所以默认不填,skip=count

结果:

1
2
3
4
5
onNext:[1, 2, 3]
onNext:[4, 5, 6]
onNext:[7, 8, 9]
onNext:[10]
onCompleted

skip值为1:

1
Observable.just(1,2,3,4,5,6,7,8,9,10).buffer(3,1).subscribe(subscriber);

结果:

1
2
3
4
5
6
7
8
9
10
11
onNext:[1, 2, 3]
onNext:[2, 3, 4]
onNext:[3, 4, 5]
onNext:[4, 5, 6]
onNext:[5, 6, 7]
onNext:[6, 7, 8]
onNext:[7, 8, 9]
onNext:[8, 9, 10]
onNext:[9, 10]
onNext:[10]
onCompleted

解释:
每次缓存了3个,再skip掉第1个,再等凑够3个,再发送,再skip掉1个,如此循环,直至结束.

filter

过滤,过滤掉不需要的数据

single

Single类似于Observable,不同的是,它总是只发射一个值,或者一个错误通知,而不是发射一系列的值。
因此,不同于Observable需要三个方法onNext, onError, onCompleted,订阅Single只需要两个方法:

  • onSuccess - Single发射单个的值到这个方法
  • onError - 如果无法发射需要的值,Single发射一个Throwable对象到这个方法

Single只会调用这两个方法中的一个,而且只会调用一次,调用了任何一个方法之后,订阅关系终止。

single 和 filter差不多,但如果为0会报错进入onError

first

tosortlist

defer

zip

merge

throttleFirst

放抖动

distinct

去重

take

doOnNext

finallyDo

contact

reduce

了解“热”和“冷”的Observable

资料/推荐

RxDocs-content
RxDocs
RxJava使用场景小结