操作符分类
操作符分类有十三种:
| 名称 |
解析 |
| 创建操作 |
用于创建Observable的操作符 |
| 变换操作 |
这些操作符可用于对Observable发射的数据进行变换 |
| 过滤操作 |
这些操作符用于从Observable发射的数据中进行选择 |
| 组合操作 |
组合操作符用于将多个Observable组合成一个单一的Observable |
| 错误处理 |
这些操作符用于从错误通知中恢复 |
| 辅助操作 |
一组用于处理Observable的操作符 |
| 条件和布尔操作 |
用于单个或多个数据项，也可用于Observable |
| 算术和聚合操作 |
用于整个数据序列 |
| 异步操作 |
单独的 rxjava-async 模块，它们用于将同步对象转换为Observable |
| 连接操作 |
一些有精确可控的订阅行为的特殊Observable |
| 转换操作 |
只有一个To，将Observable转换为其它的对象或数据结构 |
| 阻塞操作 |
阻塞Observable的操作符 |
| 字符串操作 |
一些用于处理字符串序列和流的特殊操作符 |
变换操作符
| 操作符 |
解析 |
| buffer() |
缓存，可以简单的理解为缓存，它定期从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个 |
| map() |
对序列的每一项都应用一个函数来变换Observable发射的数据序列 |
| flatMap()(无序), concatMap() (有序), flatMapIterable() |
将Observable发射的数据集合变换为Observables集合，然后将这些Observable发射的数据平坦化的放进一个单独的Observable |
| switchMap() |
将Observable发射的数据集合变换为Observables集合，然后只发射这些Observables最近发射的数据 |
| scan() |
对Observable发射的每一项数据应用一个函数，然后按顺序依次发射每一个值 |
| groupBy() |
将Observable分拆为Observable集合，将原始Observable发射的数据按Key分组，每一个Observable发射一组不同的数据 |
| buffer() |
它定期从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个 |
| window() |
定期将来自Observable的数据分拆成一些Observable窗口，然后发射这些窗口，而不是每次发射一项 |
| cast() |
在发射之前强制将Observable发射的所有数据转换为指定类型 |
过滤操作符列表
| 方法 |
含义 |
| filter() |
过滤数据 |
| takeLast() |
只发射最后的N项数据 |
| last() |
只发射最后的一项数据 |
| lastOrDefault() |
只发射最后的一项数据，如果Observable为空就发射默认值 |
| takeLastBuffer() |
将最后的N项数据当做单个数据发射 |
| skip() |
跳过开始的N项数据 |
| skipLast() |
跳过最后的N项数据 |
| take() |
只发射开始的N项数据 |
| first() , takeFirst() |
只发射第一项数据，或者满足某种条件的第一项数据 |
| firstOrDefault() |
只发射第一项数据，如果Observable为空就发射默认值 |
| elementAt() |
发射第N项数据 |
| elementAtOrDefault() |
发射第N项数据，如果Observable数据少于N项就发射默认值 |
| sample() , throttleLast() |
定期发射Observable最近的数据 |
| throttleFirst() |
定期发射Observable发射的第一项数据 |
| throttleWithTimeout() , debounce() |
只有在空闲了一段时间后才发射数据，通俗的说，就是如果一段时间没有操作，就执行一次操作 |
| timeout() |
如果在一个指定的时间段后还没发射数据，就发射一个异常 |
| distinct() |
过滤掉重复数据 |
| distinctUntilChanged() |
过滤掉连续重复的数据 |
| ofType() |
只发射指定类型的数据 |
| ignoreElements() |
丢弃所有的正常数据，只发射错误或完成通知 |
组合操作符
| 操作符 |
解析 |
| and() , then() , when() |
通过模式(And条件)和计划(Then次序)组合两个或多个Observable发射的数据集 |
| combineLatest() |
当两个Observables中的任何一个发射了一个数据时，通过一个指定的函数组合每个Observable发射的最新数据（一共两个数据），然后发射这个函数的结果 |
| join() , groupJoin() |
无论何时，如果一个Observable发射了一个数据项，只要在另一个Observable发射的数据项定义的时间窗口内，就将两个Observable发射的数据合并发射 |
| merge() |
将多个Observable合并为一个 |
| startWith() |
在数据序列的开头增加一项数据，在发射原来的Observable的数据序列之前，先发射一个指定的数据序列或数据项 |
| switch() |
将一个发射Observable序列的Observable转换为这样一个Observable：它逐个发射那些Observable最近发射的数据 |
| switchOnNext() |
将一个发射Observables的Observable转换成另一个Observable，后者发射这些Observables最近发射的数据 |
| zip() |
打包，使用一个指定的函数将多个Observable发射的数据组合在一起，然后将这个函数的结果作为单项数据发射 |
| mergeDelayError() |
合并多个Observables，让没有错误的Observable都完成后再发射错误通知 |
错误处理操作符
| 名称 |
解析 |
| onErrorResumeNext() |
指示Observable在遇到错误时发射一个数据序列 |
| onErrorReturn() |
让Observable遇到错误时发射一个特殊的项并且正常终止。方法返回一个镜像原有Observable行为的新Observable，后者会忽略前者的onError调用，不会将错误传递给观察者，作为替代，它会发发射一个特殊的项并调用观察者的onCompleted方法。 |
| onExceptionResumeNext() |
指示Observable遇到错误时继续发射数据 |
| retry() |
指示Observable遇到错误时重试 |
| retryWhen() |
指示Observable遇到错误时，将错误传递给另一个Observable来决定是否要重新给订阅这个Observable |
| retryUntil() |
指示Observable遇到错误时，是否让Observable重新订阅 |
辅助操作符
| 名称 |
解析 |
| materialize() |
将Observable转换成一个通知列表 |
| dematerialize() |
将上面的结果逆转回一个Observable |
| timestamp() |
给Observable发射的每个数据项添加一个时间戳 |
| serialize() |
强制Observable按次序发射数据并且要求功能是完好的 |
| cache() |
记住Observable发射的数据序列并发射相同的数据序列给后续的订阅者 |
| observeOn() |
指定观察者观察Observable的调度器 |
| subscribeOn() |
指定Observable执行任务的调度器 |
| doOnEach() |
注册一个动作，对Observable发射的每个数据项使用 |
| doOnCompleted() |
注册一个动作，对正常完成的Observable使用 |
| doOnError() |
注册一个动作，对发生错误的Observable使用 |
| doOnTerminate() |
Observable终止之前会被调用，无论是正常还是异常终止 |
| doOnSubscribe() |
注册一个动作，在观察者订阅时使用 |
| doOnUnsubscribe() |
注册一个动作，在观察者取消订阅时使用 |
| doOnNext() |
在onNext前执行 |
| doAfterNext() |
在onNext之后执行 |
| doAfterTerminate |
终止发送时候调用 |
| doOnLifecycle |
可以在订阅之后 设置是否取消订阅 |
| doFinally |
在最后执行 |
| finallyDo() |
注册一个动作，在Observable完成时使用 |
| delay() |
延时发射Observable的结果 |
| delaySubscription() |
延时处理订阅请求 |
| timeInterval() |
转换获取数据发送的时间间隔 |
| using() |
创建一个只在Observable生命周期存在的资源 |
| single() |
强制返回单个数据，否则抛出异常 |
| singleOrDefault() |
如果Observable完成时返回了单个数据，就返回它，否则返回默认数据 |
| toFuture(),toIterable(),toList() |
将Observable转换为其它对象或数据结构 |
条件操作符
| 名称 |
解析 |
| amb() |
给定多个Observable，只让第一个发射数据的Observable发射全部数据 |
| defaultIfEmpty() |
发射来自原始Observable的数据，如果原始Observable没有发射数据，就发射一个默认数据 |
| switchIfEmpty() |
如果原始Observable没有发射数据，它发射一个备用Observable的发射物 |
| (rxjava-computation-expressions) doWhile() |
发射原始Observable的数据序列，然后重复发射这个序列直到不满足这个条件为止 |
| (rxjava-computation-expressions) ifThen() 只有当某个条件为真时才发射原始Observable的数据序列，否则发射一个空的或默认的序列 |
| skipUntil() |
丢弃原始Observable发射的数据，直到第二个Observable发射了一个数据，然后发射原始Observable的剩余数据 |
| skipWhile() |
丢弃原始Observable发射的数据，直到一个特定的条件为假，然后发射原始Observable剩余的数据 |
| (rxjava-computation-expressions) switchCase() |
基于一个计算结果，发射一个指定Observable的数据序列 |
| takeUntil() |
发射来自原始Observable的数据，直到第二个Observable发射了一个数据或一个通知 |
| takeWhile(),takeWhileWithIndex() |
发射原始Observable的数据，直到一个特定的条件为真，然后跳过剩余的数据 |
| (rxjava-computation-expressions)whileDo() |
如果满足一个条件，发射原始Observable的数据，然后重复发射直到不满足这个条件为止 |
(rxjava-computation-expressions): 表示这个操作符当前是可选包 rxjava-
computation-expressions 的一部分，还没有包含在标准RxJava的操作符集合里,需要自己导包，但是这个包用的是Rxjava1.0.0-rc3的版本，所以在Rxjava2.0用的话，会出现问题。 可以使用repeat操作符代替
compile 'io.reactivex:rxjava-computation-expressions:0.21.0'
传送门
布尔操作符
| 名称 |
解析 |
| all() |
判断是否所有的数据项都满足某个条件 |
| contains() |
判断Observable是否会发射一个指定的值 |
| isEmpty() |
判断Observable是否发射了一个值 |
| sequenceEqual() |
判断两个Observables发射的序列是否相等 |
传送门
算术和聚合操作符
这个模块需要导下面这个包,不支持Rxjava2，所以Rxjava2用不了噢：
compile 'io.reactivex:rxjava-math:1.0.0'
| 名称 |
解析 |
| averageInteger() |
求序列平均数并发射 |
| averageLong() |
求序列平均数并发射 |
| averageFloat() |
求序列平均数并发射 |
| averageDouble() |
求序列平均数并发射 |
| max() |
求序列最大值并发射 |
| maxBy() |
求最大key对应的值并发射 |
| min() |
求最小值并发射 |
| minBy() |
求最小Key对应的值并发射 |
| sumInteger() |
求和并发射 |
| sumLong() |
求和并发射 |
| sumFloat() |
求和并发射 |
| sumDouble() |
求和并发射 |
其他聚合操作符:
| 名称 |
解析 |
| concat() / concatArray |
顺序连接多个Observables |
| count() |
计算数据项的个数并发射结果 |
| reduce() |
按顺序对Observable发射的每项数据应用一个函数并发射最终的值 |
| collect() |
将原始Observable发射的数据放到一个单一的可变的数据结构中，然后返回一个发射这个数据结构的Observable |
| toList() |
收集原始Observable发射的所有数据到一个列表，然后返回这个列表 |
| toSortedList() |
收集原始Observable发射的所有数据到一个有序列表，然后返回这个列表 |
| toMap() |
将序列数据转换为一个Map，Map的key是根据一个函数计算的 |
| toMultiMap() |
将序列数据转换为一个列表，同时也是一个Map，Map的key是根据一个函数计算的 |
ps: 由于这些操作必须等待数据发射完成（通常也必须缓存这些数据），它们对于非常长或者无限的序列来说是危险的，不推荐使用。
异步操作符
异步操作符属于单独的rxjava-async模块，它们用于将同步对象转换为Observable。不支持Rxjava2.0，如果使用Rxjava1.0的话，可以导入下面的包就可以使用异步操作符了。
compile 'io.reactivex:rxjava-async-util:0.21.0'
| 名称 |
解析 |
| start() |
创建一个Observable，它发射一个函数的返回值 |
| toAsync() / asyncAction()/ asyncFunc() |
将一个函数或者Action转换为已Observable，它执行这个函数并发射函数的返回值 |
| startFuture() |
将一个返回Future的函数转换为一个Observable，它发射Future的返回值 |
| deferFuture() |
将一个返回Observable的Future转换为一个Observable，但是并不尝试获取这个Future返回的Observable，直到有订阅者订阅它 |
| forEachFuture() |
传递Subscriber方法给一个Subscriber，但是同时表现得像一个Future一样阻塞直到它完成 |
| fromAction() |
将一个Action转换为Observable，当一个订阅者订阅时，它执行这个action并发射它的返回值 |
| fromCallable() |
将一个Callable转换为Observable，当一个订阅者订阅时，它执行这个Callable并发射Callable的返回值，或者发射异常 |
| fromRunnable() |
将一个Runnable转换为Observable，当一个订阅者订阅时，它执行这个Runnable并发射Runnable的返回值 |
| runAsync() |
返回一个StoppableObservable，它发射某个Scheduler上指定的Action生成的多个actions |
连接操作符
一个可连接的Observable与普通的Observable差不多，除了这一点：可连接的Observabe在被订阅时并不开始发射数据，只有在它的connect()被调用时才开始。用这种方法，你可以等所有的潜在订阅者都订阅了这个Observable之后才开始发射数据。
| 名称 |
解析 |
| ConnectableObservable.connect() |
指示一个可连接的Observable开始发射数据 |
| Observable.publish() |
将一个Observable转换为一个可连接的Observable |
| Observable.replay() |
确保所有的订阅者看到相同的数据序列，即使它们在Observable开始发射数据之后才订阅 |
| ConnectableObservable.refCount() |
让一个可连接的Observable表现得像一个普通的Observable |
阻塞操作符
在Rxjava1中的BlockingObservable已经在Rxjava2中去掉了，在Rxjava2中已经集成到了Observable中。
官方说明:
https://github.com/ReactiveX/RxJava/wiki/What’s-different-in-2.0
| 名称 |
解析 |
| blockingForEach() |
对Observable发射的每一项数据调用一个方法，会阻塞直到Observable完成 |
| blockingFirst() |
阻塞直到Observable发射了一个数据，然后返回第一项数据 |
| blockingMostRecent() |
返回一个总是返回Observable最近发射的数据的iterable |
| blockingLatest() |
返回一个iterable，会阻塞直到或者除非Observable发射了一个iterable没有返回的值，然后返回这个值 |
| blockingNext() |
返回一个iterable,阻塞直到返回另外一个值 |
| blockingLast() |
阻塞直到Observable终止，然后返回最后一项数据 |
| blockingIterable() |
将Observable转换返回一个iterable. |
| blockingSingle() |
如果Observable终止时只发射了一个值，返回那个值，否则抛出异常 |
| blockingSubscribe() |
在当前线程订阅，和forEach类似 |
传送门
