rxjava

Rx介绍

ReactiveX的历史

ReactiveX是Reactive Extensions的缩写，一般简写为Rx，最初是LINQ的一个扩展，由微软的架构师Erik Meijer领导的团队开发，在2012年11月开源，Rx是一个编程模型，目标是提供一致的编程接口，帮助开发者更方便的处理异步数据流，Rx库支持.NET、JavaScript和C++，Rx近几年越来越流行了，现在已经支持几乎全部的流行编程语言了，Rx的大部分语言库由ReactiveX这个组织负责维护，比较流行的有RxJava/RxJS/Rx.NET，社区网站是 reactivex.io。

什么是ReactiveX

微软给的定义是，Rx是一个函数库，让开发者可以利用可观察序列和LINQ风格查询操作符来编写异步和基于事件的程序，使用Rx，开发者可以用Observables表示异步数据流，用LINQ操作符查询异步数据流， 用Schedulers参数化异步数据流的并发处理，Rx可以这样定义：Rx = Observables + LINQ + Schedulers。

ReactiveX.io给的定义是，Rx是一个使用可观察数据流进行异步编程的编程接口，ReactiveX结合了观察者模式、迭代器模式和函数式编程的精华。

ReactiveX的应用

很多公司都在使用ReactiveX，例如Microsoft、Netflix、Github、Trello、SoundCloud。

ReactiveX宣言

ReactiveX不仅仅是一个编程接口，它是一种编程思想的突破，它影响了许多其它的程序库和框架以及编程语言。

Rx模式

使用观察者模式

• 创建：Rx可以方便的创建事件流和数据流
• 组合：Rx使用查询式的操作符组合和变换数据流
• 监听：Rx可以订阅任何可观察的数据流并执行操作

简化代码

• 函数式风格：对可观察数据流使用无副作用的输入输出函数，避免了程序里错综复杂的状态
• 简化代码：Rx的操作符通通常可以将复杂的难题简化为很少的几行代码
• 异步错误处理：传统的try/catch没办法处理异步计算，Rx提供了合适的错误处理机制
• 轻松使用并发：Rx的Observables和Schedulers让开发者可以摆脱底层的线程同步和各种并发问题

使用Observable的优势

Rx扩展了观察者模式用于支持数据和事件序列，添加了一些操作符，它让你可以声明式的组合这些序列，而无需关注底层的实现：如线程、同步、线程安全、并发数据结构和非阻塞IO。

Observable通过使用最佳的方式访问异步数据序列填补了这个间隙

Rx的Observable模型让你可以像使用集合数据一样操作异步事件流，对异步事件流使用各种简单、可组合的操作。

Observable可组合

对于单层的异步操作来说，Java中Future对象的处理方式是非常简单有效的，但是一旦涉及到嵌套，它们就开始变得异常繁琐和复杂。使用Future很难很好的组合带条件的异步执行流程（考虑到运行时各种潜在的问题，甚至可以说是不可能的），当然，要想实现还是可以做到的，但是非常困难，或许你可以用Future.get()，但这样做，异步执行的优势就完全没有了。从另一方面说，Rx的Observable一开始就是为组合异步数据流准备的。

Observable更灵活

Rx的Observable不仅支持处理单独的标量值（就像Future可以做的），也支持数据序列，甚至是无穷的数据流。Observable是一个抽象概念，适用于任何场景。Observable拥有它的近亲Iterable的全部优雅与灵活。

Observable是异步的双向push，Iterable是同步的单向pull，对比：

Observable无偏见

Rx对于对于并发性或异步性没有任何特殊的偏好，Observable可以用任何方式实现，线程池、事件循环、非阻塞IO、Actor模式，任何满足你的需求的，你擅长或偏好的方式都可以。无论你选择怎样实现它，无论底层实现是阻塞的还是非阻塞的，客户端代码将所有与Observable的交互都当做是异步的。

Observable是如何实现的？

public Observable<data> getData();

• 它能与调用者在同一线程同步执行吗？
• 它能异步地在单独的线程执行吗？
• 它会将工作分发到多个线程，返回数据的顺序是任意的吗？
• 它使用Actor模式而不是线程池吗？
• 它使用NIO和事件循环执行异步网络访问吗？
• 它使用事件循环将工作线程从回调线程分离出来吗？

从Observer的视角看，这些都无所谓，重要的是：使用Rx，你可以改变你的观念，你可以在完全不影响Observable程序库使用者的情况下，彻底的改变Observable的底层实现。

使用回调存在很多问题

回调在不阻塞任何事情的情况下，解决了Future.get()过早阻塞的问题。由于响应结果一旦就绪Callback就会被调用，它们天生就是高效率的。不过，就像使用Future一样，对于单层的异步执行来说，回调很容易使用，对于嵌套的异步组合，它们显得非常笨拙。

Rx是一个多语言的实现

Rx在大量的编程语言中都有实现，并尊重实现语言的风格，而且更多的实现正在飞速增加。

响应式编程

Rx提供了一系列的操作符，你可以使用它们来过滤(filter)、选择(select)、变换(transform)、结合(combine)和组合(compose)多个Observable，这些操作符让执行和复合变得非常高效。

你可以把Observable当做Iterable的推送方式的等价物，使用Iterable，消费者从生产者那拉取数据，线程阻塞直至数据准备好。使用Observable，在数据准备好时，生产者将数据推送给消费者。数据可以同步或异步的到达，这种方式更灵活。

下面的例子展示了相似的高阶函数在Iterable和Observable上的应用

// Iterable
getDataFromLocalMemory()
  .skip(10)
  .take(5)
  .map({ s -> return s + " transformed" })
  .forEach({ println "next => " + it })

// Observable
getDataFromNetwork()
  .skip(10)
  .take(5)
  .map({ s -> return s + " transformed" })
  .subscribe({ println "onNext => " + it })

Observable类型给GOF的观察者模式添加了两种缺少的语义，这样就和Iterable类型中可用的操作一致了：

1. 生产者可以发信号给消费者，通知它没有更多数据可用了（对于Iterable，一个for循环正常完成表示没有数据了；对于Observable，就是调用观察者的onCompleted方法）
2. 生产者可以发信号给消费者，通知它遇到了一个错误（对于Iterable，迭代过程中发生错误会抛出异常；对于Observable，就是调用观察者(Observer)的onError方法）

有了这两种功能，Rx就能使Observable与Iterable保持一致了，唯一的不同是数据流的方向。任何对Iterable的操作，你都可以对Observable使用。

名词定义

这里给出一些名词的翻译

• Reactive 直译为反应性的，有活性的，根据上下文一般翻译为反应式、响应式
• Iterable 可迭代对象，支持以迭代器的形式遍历，许多语言中都存在这个概念
• Observable 可观察对象，在Rx中定义为更强大的Iterable，在观察者模式中是被观察的对象，一旦数据产生或发生变化，会通过某种方式通知观察者或订阅者
• Observer 观察者对象，监听Observable发射的数据并做出响应，Subscriber是它的一个特殊实现
• emit 直译为发射，发布，发出，含义是Observable在数据产生或变化时发送通知给Observer，调用Observer对应的方法，文章里一律译为发射
• items 直译为项目，条目，在Rx里是指Observable发射的数据项，文章里一律译为数据，数据项

Observable

概述

在ReactiveX中，一个观察者(Observer)订阅一个可观察对象(Observable)。观察者对Observable发射的数据或数据序列作出响应。这种模式可以极大地简化并发操作，因为它创建了一个处于待命状态的观察者哨兵，在未来某个时刻响应Observable的通知，不需要阻塞等待Observable发射数据。

这篇文章会解释什么是响应式编程模式(reactive pattern)，以及什么是可观察对象(Observables)和观察者(observers)，其它几篇文章会展示如何用操作符组合和改变Observable的行为。

Observable

相关参考：

• Single - 一个特殊的Observable，只发射单个数据。

背景知识

在很多软件编程任务中，或多或少你都会期望你写的代码能按照编写的顺序，一次一个的顺序执行和完成。但是在ReactiveX中，很多指令可能是并行执行的，之后他们的执行结果才会被观察者捕获，顺序是不确定的。为达到这个目的，你定义一种获取和变换数据的机制，而不是调用一个方法。在这种机制下，存在一个可观察对象(Observable)，观察者(Observer)订阅(Subscribe)它，当数据就绪时，之前定义的机制就会分发数据给一直处于等待状态的观察者哨兵。

这种方法的优点是，如果你有大量的任务要处理，它们互相之间没有依赖关系。你可以同时开始执行它们，不用等待一个完成再开始下一个（用这种方式，你的整个任务队列能耗费的最长时间，不会超过任务里最耗时的那个）。

有很多术语可用于描述这种异步编程和设计模式，在在本文里我们使用这些术语：一个观察者订阅一个可观察对象 (An observer subscribes to an Observable)。通过调用观察者的方法，Observable发射数据或通知给它的观察者。

在其它的文档和场景里，有时我们也将Observer叫做Subscriber、Watcher、Reactor。这个模型通常被称作Reactor模式。

创建观察者

本文使用类似于Groovy的伪代码举例，但是ReactiveX有多种语言的实现。

普通的方法调用（不是某种异步方法，也不是Rx中的并行调用），流程通常是这样的：

1. 调用某一个方法
2. 用一个变量保存方法返回的结果
3. 使用这个变量和它的新值做些有用的事

用代码描述就是：

// make the call, assign its return value to `returnVal`
returnVal = someMethod(itsParameters);
// do something useful with returnVal

在异步模型中流程更像这样的：

1. 定义一个方法，这个方法拿着某个异步调用的返回值做一些有用的事情。这个方法是观察者的一部分。
2. 将这个异步调用本身定义为一个Observable
3. 观察者通过订阅(Subscribe)操作关联到那个Observable
4. 继续你的业务逻辑，等方法返回时，Observable会发射结果，观察者的方法会开始处理结果或结果集

用代码描述就是：

// defines, but does not invoke, the Subscriber's onNext handler
// (in this example, the observer is very simple and has only an onNext handler)
def myOnNext = { it -> do something useful with it };
// defines, but does not invoke, the Observable
def myObservable = someObservable(itsParameters);
// subscribes the Subscriber to the Observable, and invokes the Observable
myObservable.subscribe(myOnNext);
// go on about my business

回调方法 (onNext, onCompleted, onError)

Subscribe方法用于将观察者连接到Observable，你的观察者需要实现以下方法的一个子集：

• onNext(T item)

  Observable调用这个方法发射数据，方法的参数就是Observable发射的数据，这个方法可能会被调用多次，取决于你的实现。

• onError(Exception ex)

  当Observable遇到错误或者无法返回期望的数据时会调用这个方法，这个调用会终止Observable，后续不会再调用onNext和onCompleted，onError方法的参数是抛出的异常。

• onComplete

  正常终止，如果没有遇到错误，Observable在最后一次调用onNext之后调用此方法。

根据Observable协议的定义，onNext可能会被调用零次或者很多次，最后会有一次onCompleted或onError调用（不会同时），传递数据给onNext通常被称作发射，onCompleted和onError被称作通知。

下面是一个更完整的例子：

def myOnNext     = { item -> /* do something useful with item */ };
def myError      = { throwable -> /* react sensibly to a failed call */ };
def myComplete   = { /* clean up after the final response */ };
def myObservable = someMethod(itsParameters);
myObservable.subscribe(myOnNext, myError, myComplete);
// go on about my business

取消订阅 (Unsubscribing)

在一些ReactiveX实现中，有一个特殊的观察者接口Subscriber，它有一个unsubscribe方法。调用这个方法表示你不关心当前订阅的Observable了，因此Observable可以选择停止发射新的数据项（如果没有其它观察者订阅）。

取消订阅的结果会传递给这个Observable的操作符链，而且会导致这个链条上的每个环节都停止发射数据项。这些并不保证会立即发生，然而，对一个Observable来说，即使没有观察者了，它也可以在一个while循环中继续生成并尝试发射数据项。

关于命名约定

ReactiveX的每种特定语言的实现都有自己的命名偏好，虽然不同的实现之间有很多共同点，但并不存在一个统一的命名标准。

而且，在某些场景中，一些名字有不同的隐含意义，或者在某些语言看来比较怪异。

例如，有一个onEvent命名模式(onNext, onCompleted, onError)，在一些场景中，这些名字可能意味着事件处理器已经注册。然而在ReactiveX里，他们是事件处理器的名字。

Observables的"热"和"冷"

Observable什么时候开始发射数据序列？这取决于Observable的实现，一个"热"的Observable可能一创建完就开始发射数据，因此所有后续订阅它的观察者可能从序列中间的某个位置开始接受数据（有一些数据错过了）。一个"冷"的Observable会一直等待，直到有观察者订阅它才开始发射数据，因此这个观察者可以确保会收到整个数据序列。

在一些ReactiveX实现里，还存在一种被称作Connectable的Observable，不管有没有观察者订阅它，这种Observable都不会开始发射数据，除非Connect方法被调用。

用操作符组合Observable

对于ReactiveX来说，Observable和Observer仅仅是个开始，它们本身不过是标准观察者模式的一些轻量级扩展，目的是为了更好的处理事件序列。

ReactiveX真正强大的地方在于它的操作符，操作符让你可以变换、组合、操纵和处理Observable发射的数据。

Rx的操作符让你可以用声明式的风格组合异步操作序列，它拥有回调的所有效率优势，同时又避免了典型的异步系统中嵌套回调的缺点。

下面是常用的操作符列表：

1. 创建操作 Create, Defer, Empty/Never/Throw, From, Interval, Just, Range, Repeat, Start, Timer
2. 变换操作 Buffer, FlatMap, GroupBy, Map, Scan和Window
3. 过滤操作 Debounce, Distinct, ElementAt, Filter, First, IgnoreElements, Last, Sample, Skip, SkipLast, Take, TakeLast
4. 组合操作 And/Then/When, CombineLatest, Join, Merge, StartWith, Switch, Zip
5. 错误处理 Catch和Retry
6. 辅助操作 Delay, Do, Materialize/Dematerialize, ObserveOn, Serialize, Subscribe, SubscribeOn, TimeInterval, Timeout, Timestamp, Using
7. 条件和布尔操作 All, Amb, Contains, DefaultIfEmpty, SequenceEqual, SkipUntil, SkipWhile, TakeUntil, TakeWhile
8. 算术和集合操作 Average, Concat, Count, Max, Min, Reduce, Sum
9. 转换操作 To
10. 连接操作 Connect, Publish, RefCount, Replay
11. 反压操作，用于增加特殊的流程控制策略的操作符

这些操作符并不全都是ReactiveX的核心组成部分，有一些是语言特定的实现或可选的模块。

RxJava

在RxJava中，一个实现了Observer接口的对象可以订阅(subscribe)一个Observable 类的实例。订阅者(subscriber)对Observable发射(emit)的任何数据或数据序列作出响应。这种模式简化了并发操作，因为它不需要阻塞等待Observable发射数据，而是创建了一个处于待命状态的观察者哨兵，哨兵在未来某个时刻响应Observable的通知。

Single

介绍

RxJava（以及它派生出来的RxGroovy和RxScala）中有一个名为Single的Observable变种。

Single类似于Observable，不同的是，它总是只发射一个值，或者一个错误通知，而不是发射一系列的值。

因此，不同于Observable需要三个方法onNext, onError, onCompleted，订阅Single只需要两个方法：

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

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

Single的操作符

Single也可以组合使用多种操作，一些操作符让你可以混合使用Observable和Single：

操作符图示

详细的图解可以参考英文文档：Single

Subject

Subject可以看成是一个桥梁或者代理，在某些ReactiveX实现中（如RxJava），它同时充当了Observer和Observable的角色。因为它是一个Observer，它可以订阅一个或多个Observable；又因为它是一个Observable，它可以转发它收到(Observe)的数据，也可以发射新的数据。

由于一个Subject订阅一个Observable，它可以触发这个Observable开始发射数据（如果那个Observable是"冷"的--就是说，它等待有订阅才开始发射数据）。因此有这样的效果，Subject可以把原来那个"冷"的Observable变成"热"的。

Subject的种类

针对不同的场景一共有四种类型的Subject。他们并不是在所有的实现中全部都存在，而且一些实现使用其它的命名约定（例如，在RxScala中Subject被称作PublishSubject）。

AsyncSubject

img

img

BehaviorSubject

img

img

PublishSubject

img

img

ReplaySubject

ReplaySubject会发射所有来自原始Observable的数据给观察者，无论它们是何时订阅的。也有其它版本的ReplaySubject，在重放缓存增长到一定大小的时候或过了一段时间后会丢弃旧的数据（原始Observable发射的）。

如果你把ReplaySubject当作一个观察者使用，注意不要从多个线程中调用它的onNext方法（包括其它的on系列方法），这可能导致同时（非顺序）调用，这会违反Observable协议，给Subject的结果增加了不确定性。

img

RxJava的对应类

假设你有一个Subject，你想把它传递给其它的代理或者暴露它的Subscriber接口，你可以调用它的asObservable方法，这个方法返回一个Observable。具体使用方法可以参考Javadoc文档。

串行化

如果你把 Subject 当作一个 Subscriber 使用，注意不要从多个线程中调用它的onNext方法（包括其它的on系列方法），这可能导致同时（非顺序）调用，这会违反Observable协议，给Subject的结果增加了不确定性。

要避免此类问题，你可以将 Subject 转换为一个 SerializedSubject ，类似于这样：

mySafeSubject = new SerializedSubject( myUnsafeSubject );

调度器 Scheduler

如果你想给Observable操作符链添加多线程功能，你可以指定操作符（或者特定的Observable）在特定的调度器(Scheduler)上执行。

某些ReactiveX的Observable操作符有一些变体，它们可以接受一个Scheduler参数。这个参数指定操作符将它们的部分或全部任务放在一个特定的调度器上执行。

使用ObserveOn和SubscribeOn操作符，你可以让Observable在一个特定的调度器上执行，ObserveOn指示一个Observable在一个特定的调度器上调用观察者的onNext, onError和onCompleted方法，SubscribeOn更进一步，它指示Observable将全部的处理过程（包括发射数据和通知）放在特定的调度器上执行。

RxJava示例

调度器的种类

下表展示了RxJava中可用的调度器种类：

默认调度器

在RxJava中，某些Observable操作符的变体允许你设置用于操作执行的调度器，其它的则不在任何特定的调度器上执行，或者在一个指定的默认调度器上执行。下面的表格个列出了一些操作符的默认调度器：

使用调度器

除了将这些调度器传递给RxJava的Observable操作符，你也可以用它们调度你自己的任务。下面的示例展示了Scheduler.Worker的用法：

worker = Schedulers.newThread().createWorker();
worker.schedule(new Action0() {

    @Override
    public void call() {
        yourWork();
    }

});
// some time later...
worker.unsubscribe();

递归调度器

要调度递归的方法调用，你可以使用schedule，然后再用schedule(this)，示例：

worker = Schedulers.newThread().createWorker();
worker.schedule(new Action0() {

    @Override
    public void call() {
        yourWork();
        // recurse until unsubscribed (schedule will do nothing if unsubscribed)
        worker.schedule(this);
    }

});
// some time later...
worker.unsubscribe();

检查或设置取消订阅状态

Worker类的对象实现了Subscription接口，使用它的isUnsubscribed和unsubscribe方法，所以你可以在订阅取消时停止任务，或者从正在调度的任务内部取消订阅，示例：

Worker worker = Schedulers.newThread().createWorker();
Subscription mySubscription = worker.schedule(new Action0() {

    @Override
    public void call() {
        while(!worker.isUnsubscribed()) {
            status = yourWork();
            if(QUIT == status) { worker.unsubscribe(); }
        }
    }

});

Worker同时是Subscription，因此你可以（通常也应该）调用它的unsubscribe方法通知可以挂起任务和释放资源了。

延时和周期调度器

你可以使用schedule(action,delayTime,timeUnit)在指定的调度器上延时执行你的任务，下面例子中的任务将在500毫秒之后开始执行：

someScheduler.schedule(someAction, 500, TimeUnit.MILLISECONDS);

使用另一个版本的schedule，schedulePeriodically(action,initialDelay,period,timeUnit)方法让你可以安排一个定期执行的任务，下面例子的任务将在500毫秒之后执行，然后每250毫秒执行一次：

someScheduler.schedulePeriodically(someAction, 500, 250, TimeUnit.MILLISECONDS);

测试调度器

TestScheduler让你可以对调度器的时钟表现进行手动微调。这对依赖精确时间安排的任务的测试很有用处。这个调度器有三个额外的方法：

• advanceTimeTo(time,unit) 向前波动调度器的时钟到一个指定的时间点
• advanceTimeBy(time,unit) 将调度器的时钟向前拨动一个指定的时间段
• triggerActions( ) 开始执行任何计划中的但是未启动的任务，如果它们的计划时间等于或者早于调度器时钟的当前时间

操作符分类

ReactiveX的每种编程语言的实现都实现了一组操作符的集合。不同的实现之间有很多重叠的部分，也有一些操作符只存在特定的实现中。每种实现都倾向于用那种编程语言中他们熟悉的上下文中相似的方法给这些操作符命名。

本文首先会给出ReactiveX的核心操作符列表和对应的文档链接，后面还有一个决策树用于帮助你根据具体的场景选择合适的操作符。最后有一个语言特定实现的按字母排序的操作符列表。

如果你想实现你自己的操作符，可以参考这里：实现自定义操作符

创建操作

用于创建Observable的操作符

• Create — 通过调用观察者的方法从头创建一个Observable
• Defer — 在观察者订阅之前不创建这个Observable，为每一个观察者创建一个新的Observable
• Empty/Never/Throw — 创建行为受限的特殊Observable
• From — 将其它的对象或数据结构转换为Observable
• Interval — 创建一个定时发射整数序列的Observable
• Just — 将对象或者对象集合转换为一个会发射这些对象的Observable
• Range — 创建发射指定范围的整数序列的Observable
• Repeat — 创建重复发射特定的数据或数据序列的Observable
• Start — 创建发射一个函数的返回值的Observable
• Timer — 创建在一个指定的延迟之后发射单个数据的Observable

变换操作

这些操作符可用于对Observable发射的数据进行变换，详细解释可以看每个操作符的文档

• Buffer — 缓存，可以简单的理解为缓存，它定期从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个
• FlatMap — 扁平映射，将Observable发射的数据变换为Observables集合，然后将这些Observable发射的数据平坦化的放进一个单独的Observable，可以认为是一个将嵌套的数据结构展开的过程。
• GroupBy — 分组，将原来的Observable分拆为Observable集合，将原始Observable发射的数据按Key分组，每一个Observable发射一组不同的数据
• Map — 映射，通过对序列的每一项都应用一个函数变换Observable发射的数据，实质是对序列中的每一项执行一个函数，函数的参数就是这个数据项
• Scan — 扫描，对Observable发射的每一项数据应用一个函数，然后按顺序依次发射这些值
• Window — 窗口，定期将来自Observable的数据分拆成一些Observable窗口，然后发射这些窗口，而不是每次发射一项。类似于Buffer，但Buffer发射的是数据，Window发射的是Observable，每一个Observable发射原始Observable的数据的一个子集

过滤操作

这些操作符用于从Observable发射的数据中进行选择

• Debounce — 只有在空闲了一段时间后才发射数据，通俗的说，就是如果一段时间没有操作，就执行一次操作
• Distinct — 去重，过滤掉重复数据项
• ElementAt — 取值，取特定位置的数据项
• Filter — 过滤，过滤掉没有通过谓词测试的数据项，只发射通过测试的
• First — 首项，只发射满足条件的第一条数据
• IgnoreElements — 忽略所有的数据，只保留终止通知(onError或onCompleted)
• Last — 末项，只发射最后一条数据
• Sample — 取样，定期发射最新的数据，等于是数据抽样，有的实现里叫ThrottleFirst
• Skip — 跳过前面的若干项数据
• SkipLast — 跳过后面的若干项数据
• Take — 只保留前面的若干项数据
• TakeLast — 只保留后面的若干项数据

组合操作

组合操作符用于将多个Observable组合成一个单一的Observable

• And/Then/When — 通过模式(And条件)和计划(Then次序)组合两个或多个Observable发射的数据集
• CombineLatest — 当两个Observables中的任何一个发射了一个数据时，通过一个指定的函数组合每个Observable发射的最新数据（一共两个数据），然后发射这个函数的结果
• Join — 无论何时，如果一个Observable发射了一个数据项，只要在另一个Observable发射的数据项定义的时间窗口内，就将两个Observable发射的数据合并发射
• Merge — 将两个Observable发射的数据组合并成一个
• StartWith — 在发射原来的Observable的数据序列之前，先发射一个指定的数据序列或数据项
• Switch — 将一个发射Observable序列的Observable转换为这样一个Observable：它逐个发射那些Observable最近发射的数据
• Zip — 打包，使用一个指定的函数将多个Observable发射的数据组合在一起，然后将这个函数的结果作为单项数据发射

错误处理

这些操作符用于从错误通知中恢复

• Catch — 捕获，继续序列操作，将错误替换为正常的数据，从onError通知中恢复
• Retry — 重试，如果Observable发射了一个错误通知，重新订阅它，期待它正常终止

辅助操作

一组用于处理Observable的操作符

• Delay — 延迟一段时间发射结果数据
• Do — 注册一个动作占用一些Observable的生命周期事件，相当于Mock某个操作
• Materialize/Dematerialize — 将发射的数据和通知都当做数据发射，或者反过来
• ObserveOn — 指定观察者观察Observable的调度程序（工作线程）
• Serialize — 强制Observable按次序发射数据并且功能是有效的
• Subscribe — 收到Observable发射的数据和通知后执行的操作
• SubscribeOn — 指定Observable应该在哪个调度程序上执行
• TimeInterval — 将一个Observable转换为发射两个数据之间所耗费时间的Observable
• Timeout — 添加超时机制，如果过了指定的一段时间没有发射数据，就发射一个错误通知
• Timestamp — 给Observable发射的每个数据项添加一个时间戳
• Using — 创建一个只在Observable的生命周期内存在的一次性资源

条件和布尔操作

这些操作符可用于单个或多个数据项，也可用于Observable

• All — 判断Observable发射的所有的数据项是否都满足某个条件
• Amb — 给定多个Observable，只让第一个发射数据的Observable发射全部数据
• Contains — 判断Observable是否会发射一个指定的数据项
• DefaultIfEmpty — 发射来自原始Observable的数据，如果原始Observable没有发射数据，就发射一个默认数据
• SequenceEqual — 判断两个Observable是否按相同的数据序列
• SkipUntil — 丢弃原始Observable发射的数据，直到第二个Observable发射了一个数据，然后发射原始Observable的剩余数据
• SkipWhile — 丢弃原始Observable发射的数据，直到一个特定的条件为假，然后发射原始Observable剩余的数据
• TakeUntil — 发射来自原始Observable的数据，直到第二个Observable发射了一个数据或一个通知
• TakeWhile — 发射原始Observable的数据，直到一个特定的条件为真，然后跳过剩余的数据

算术和聚合操作

这些操作符可用于整个数据序列

• Average — 计算Observable发射的数据序列的平均值，然后发射这个结果
• Concat — 不交错的连接多个Observable的数据
• Count — 计算Observable发射的数据个数，然后发射这个结果
• Max — 计算并发射数据序列的最大值
• Min — 计算并发射数据序列的最小值
• Reduce — 按顺序对数据序列的每一个应用某个函数，然后返回这个值
• Sum — 计算并发射数据序列的和

连接操作

一些有精确可控的订阅行为的特殊Observable

• Connect — 指示一个可连接的Observable开始发射数据给订阅者
• Publish — 将一个普通的Observable转换为可连接的
• RefCount — 使一个可连接的Observable表现得像一个普通的Observable
• Replay — 确保所有的观察者收到同样的数据序列，即使他们在Observable开始发射数据之后才订阅

转换操作

• To — 将Observable转换为其它的对象或数据结构
• Blocking 阻塞Observable的操作符

操作符决策树

几种主要的需求

• 直接创建一个Observable（创建操作）
• 组合多个Observable（组合操作）
• 对Observable发射的数据执行变换操作（变换操作）
• 从Observable发射的数据中取特定的值（过滤操作）
• 转发Observable的部分值（条件/布尔/过滤操作）
• 对Observable发射的数据序列求值（算术/聚合操作）

这个页面展示了创建Observable的各种方法。

• just( ) — 将一个或多个对象转换成发射这个或这些对象的一个Observable
• from( ) — 将一个Iterable, 一个Future, 或者一个数组转换成一个Observable
• repeat( ) — 创建一个重复发射指定数据或数据序列的Observable
• repeatWhen( ) — 创建一个重复发射指定数据或数据序列的Observable，它依赖于另一个Observable发射的数据
• create( ) — 使用一个函数从头创建一个Observable
• defer( ) — 只有当订阅者订阅才创建Observable；为每个订阅创建一个新的Observable
• range( ) — 创建一个发射指定范围的整数序列的Observable
• interval( ) — 创建一个按照给定的时间间隔发射整数序列的Observable
• timer( ) — 创建一个在给定的延时之后发射单个数据的Observable
• empty( ) — 创建一个什么都不做直接通知完成的Observable
• error( ) — 创建一个什么都不做直接通知错误的Observable
• never( ) — 创建一个不发射任何数据的Observable

创建操作

Create

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

create

你可以使用Create操作符从头开始创建一个Observable，给这个操作符传递一个接受观察者作为参数的函数，编写这个函数让它的行为表现为一个Observable--恰当的调用观察者的onNext，onError和onCompleted方法。

一个形式正确的有限Observable必须尝试调用观察者的onCompleted正好一次或者它的onError正好一次，而且此后不能再调用观察者的任何其它方法。

create

RxJava将这个操作符实现为 create 方法。

建议你在传递给create方法的函数中检查观察者的isUnsubscribed状态，以便在没有观察者的时候，让你的Observable停止发射数据或者做昂贵的运算。

示例代码：

Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> observer) {
        try {
            if (!observer.isUnsubscribed()) {
                for (int i = 1; i < 5; i++) {
                    observer.onNext(i);
                }
                observer.onCompleted();
            }
        } catch (Exception e) {
            observer.onError(e);
        }
    }
 } ).subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }

        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });

输出：

Next: 1
Next: 2
Next: 3
Next: 4
Sequence complete.

create方法默认不在任何特定的调度器上执行。

• Javadoc: create(OnSubscribe)

Defer

直到有观察者订阅时才创建Observable，并且为每个观察者创建一个新的Observable

defer

Defer操作符会一直等待直到有观察者订阅它，然后它使用Observable工厂方法生成一个Observable。它对每个观察者都这样做，因此尽管每个订阅者都以为自己订阅的是同一个Observable，事实上每个订阅者获取的是它们自己的单独的数据序列。

在某些情况下，等待直到最后一分钟（就是知道订阅发生时）才生成Observable可以确保Observable包含最新的数据。

defer

RxJava将这个操作符实现为 defer 方法。这个操作符接受一个你选择的Observable工厂函数作为单个参数。这个函数没有参数，返回一个Observable。

defer方法默认不在任何特定的调度器上执行。

• Javadoc: defer(Func0)

switchCase

switchCase

可选包 rxjava-computation-expressions 中有一个类似的操作符。switchCase操作符有条件的创建并返回一个可能的Observables集合中的一个。

ifThen

可选包 rxjava-computation-expressions 中还有一个更简单的操作符叫ifThen。这个操作符检查某个条件，然后根据结果，返回原始Observable的镜像，或者返回一个空Observable。

Empty/Never/Throw

Empty

创建一个不发射任何数据但是正常终止的Observable

Never

创建一个不发射数据也不终止的Observable

Throw

创建一个不发射数据以一个错误终止的Observable

这三个操作符生成的Observable行为非常特殊和受限。测试的时候很有用，有时候也用于结合其它的Observables，或者作为其它需要Observable的操作符的参数。

RxJava将这些操作符实现为 empty，never和error。error操作符需要一个Throwable参数，你的Observable会以此终止。这些操作符默认不在任何特定的调度器上执行，但是empty和error有一个可选参数是Scheduler，如果你传递了Scheduler参数，它们会在这个调度器上发送通知。

• Javadoc: empty()
• Javadoc: never()
• Javadoc: error(java.lang.Throwable)

From

将其它种类的对象和数据类型转换为Observable

from

当你使用Observable时，如果你要处理的数据都可以转换成展现为Observables，而不是需要混合使用Observables和其它类型的数据，会非常方便。这让你在数据流的整个生命周期中，可以使用一组统一的操作符来管理它们。

例如，Iterable可以看成是同步的Observable；Future，可以看成是总是只发射单个数据的Observable。通过显式地将那些数据转换为Observables，你可以像使用Observable一样与它们交互。

因此，大部分ReactiveX实现都提供了将语言特定的对象和数据结构转换为Observables的方法。

from

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

示例代码

Integer[] items = { 0, 1, 2, 3, 4, 5 };
Observable myObservable = Observable.from(items);

myObservable.subscribe(
    new Action1<Integer>() {
        @Override
        public void call(Integer item) {
            System.out.println(item);
        }
    },
    new Action1<Throwable>() {
        @Override
        public void call(Throwable error) {
            System.out.println("Error encountered: " + error.getMessage());
        }
    },
    new Action0() {
        @Override
        public void call() {
            System.out.println("Sequence complete");
        }
    }
);

输出

0
1
2
3
4
5
Sequence complete

对于Future，它会发射Future.get()方法返回的单个数据。from方法有一个可接受两个可选参数的版本，分别指定超时时长和时间单位。如果过了指定的时长Future还没有返回一个值，这个Observable会发射错误通知并终止。

from默认不在任何特定的调度器上执行。然而你可以将Scheduler作为可选的第二个参数传递给Observable，它会在那个调度器上管理这个Future。

• Javadoc: from(array)
• Javadoc: from(Iterable)
• Javadoc: from(Future)
• Javadoc: from(Future,Scheduler)
• Javadoc: from(Future,timeout, timeUnit)

RxJavaAsyncUtil

from func)

此外，在可选包 RxJavaAsyncUtil 中，你还可以用下面这些操作符将actions，callables，functions和runnables转换为发射这些动作的执行结果的Observable：

• fromAction
• fromCallable
• fromFunc0
• fromRunnable

在这个页面 Start 查看关于这些操作符的更多信息。

from

注意：还有一个可选的StringObservable类中也有一个from方法，它将一个字符流或者一个REader转换为一个发射字节数组或字符串的Observable。

runAsync2

注意：这里与后面start操作符里的runAsync说明重复了

在单独的RxJavaAsyncUtil包中（默认不包含在RxJava中）,还有一个runAsync函数。传递一个Action和一个Scheduler给runAsync，它会返回一个StoppableObservable，这个Observable使用Action产生发射的数据项。

传递一个Action和一个Scheduler给runAsync，它返回一个使用这个Action产生数据的StoppableObservable。这个Action接受一个Observable和一个Subscription作为参数，它使用Subscription检查unsubscribed条件，一旦发现条件为真就立即停止发射数据。在任何时候你都可以使用unsubscribe方法手动停止一个StoppableObservable（这会同时取消订阅与这个StoppableObservable关联的Subscription）。

由于runAsync会立即调用Action并开始发射数据，在你创建StoppableObservable之后到你的观察者准备好接受数据之前这段时间里，可能会有一部分数据会丢失。如果这不符合你的要求，可以使用runAsync的一个变体，它也接受一个Subject参数，传递一个ReplaySubject给它，你可以获取其它丢失的数据了。

decode

decode

StringObservable类不是默认RxJava的一部分，包含一个decode操作符，这个操作符将一个多字节字符流转换为一个发射字节数组的Observable，这些字节数组按照字符的边界划分。

Interval

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

interval

Interval操作符返回一个Observable，它按固定的时间间隔发射一个无限递增的整数序列。

interval

RxJava将这个操作符实现为interval方法。它接受一个表示时间间隔的参数和一个表示时间单位的参数。

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

timer

还有一个版本的interval返回一个Observable，它在指定延迟之后先发射一个零值，然后再按照指定的时间间隔发射递增的数字。这个版本的interval在RxJava 1.0.0中叫做timer，但是那个方法已经不建议使用了，因为一个名叫interval的操作符有同样的功能。

Javadoc: interval(long,long,TimeUnit) Javadoc: interval(long,long,TimeUnit,Scheduler)

interval默认在computation调度器上执行。你也可以传递一个可选的Scheduler参数来指定调度器。

Just

创建一个发射指定值的Observable

just

Just将单个数据转换为发射那个数据的Observable。

Just类似于From，但是From会将数组或Iterable的数据取出然后逐个发射，而Just只是简单的原样发射，将数组或Iterable当做单个数据。

注意：如果你传递null给Just，它会返回一个发射null值的Observable。不要误认为它会返回一个空Observable（完全不发射任何数据的Observable），如果需要空Observable你应该使用Empty操作符。

RxJava将这个操作符实现为just函数，它接受一至九个参数，返回一个按参数列表顺序发射这些数据的Observable。

示例代码：

Observable.just(1, 2, 3)
          .subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }

        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });

输出

Next: 1
Next: 2
Next: 3
Sequence complete.

• Javadoc: just(item) (还有其它接受二到九个参数的版本)

Range

创建一个发射特定整数序列的Observable

range

Range操作符发射一个范围内的有序整数序列，你可以指定范围的起始和长度。

RxJava将这个操作符实现为range函数，它接受两个参数，一个是范围的起始值，一个是范围的数据的数目。如果你将第二个参数设为0，将导致Observable不发射任何数据（如果设置为负数，会抛异常）。

range默认不在任何特定的调度器上执行。有一个变体可以通过可选参数指定Scheduler。

• Javadoc: range(int,int)
• Javadoc: range(int,int,Scheduler)

Repeat

创建一个发射特定数据重复多次的Observable

repeat

Repeat重复地发射数据。某些实现允许你重复的发射某个数据序列，还有一些允许你限制重复的次数。

repeat

RxJava将这个操作符实现为repeat方法。它不是创建一个Observable，而是重复发射原始Observable的数据序列，这个序列或者是无限的，或者通过repeat(n)指定重复次数。

repeat操作符默认在trampoline调度器上执行。有一个变体可以通过可选参数指定Scheduler。

Javadoc: repeat() Javadoc: repeat(long) Javadoc: repeat(Scheduler) Javadoc: repeat(long,Scheduler)

repeatWhen

repeatWhen

还有一个叫做repeatWhen的操作符，它不是缓存和重放原始Observable的数据序列，而是有条件的重新订阅和发射原来的Observable。

将原始Observable的终止通知（完成或错误）当做一个void数据传递给一个通知处理器，它以此来决定是否要重新订阅和发射原来的Observable。这个通知处理器就像一个Observable操作符，接受一个发射void通知的Observable为输入，返回一个发射void数据（意思是，重新订阅和发射原始Observable）或者直接终止（意思是，使用repeatWhen终止发射数据）的Observable。

repeatWhen操作符默认在trampoline调度器上执行。有一个变体可以通过可选参数指定Scheduler。

• Javadoc: repeatWhen(Func1)
• Javadoc: repeatWhen(Func1,Scheduler)

doWhile

doWhile

doWhile 属于可选包rxjava-computation-expressions，不是RxJava标准操作符的一部分。doWhile在原始序列的每次重复后检查某个条件，如果满足条件才重复发射。

whileDo

whileDo

whileDo 属于可选包rxjava-computation-expressions，不是RxJava标准操作符的一部分。whileDo在原始序列的每次重复前检查某个条件，如果满足条件才重复发射

Start

返回一个Observable，它发射一个类似于函数声明的值

Start

编程语言有很多种方法可以从运算结果中获取值，它们的名字一般叫functions, futures, actions, callables, runnables等等。在Start目录下的这组操作符可以让它们表现得像Observable，因此它们可以在Observables调用链中与其它Observable搭配使用。

Start操作符的多种RxJava实现都属于可选的rxjava-async模块。

rxjava-async模块包含start操作符，它接受一个函数作为参数，调用这个函数获取一个值，然后返回一个会发射这个值给后续观察者的Observable。

注意：这个函数只会被执行一次，即使多个观察者订阅这个返回的Observable。

toAsync

toAsync

rxjava-async模块还包含这几个操作符：toAsync, asyncAction, 和asyncFunc。它们接受一个函数或一个Action作为参数。

对于函数(functions)，这个操作符调用这个函数获取一个值，然后返回一个会发射这个值给后续观察者的Observable（和start一样）。对于动作(Action)，过程类似，但是没有返回值，在这种情况下，这个操作符在终止前会发射一个null值。

注意：这个函数或动作只会被执行一次，即使多个观察者订阅这个返回的Observable。

startFuture

startFuture

rxjava-async模块还包含一个startFuture操作符，传递给它一个返回Future的函数，startFuture会立即调用这个函数获取Future对象，然后调用Future的get()方法尝试获取它的值。它返回一个发射这个值给后续观察者的Observable。

deferFuture

deferFuture

rxjava-async模块还包含一个deferFuture操作符，传递给它一个返回Future的函数（这个Future返回一个Observable），deferFuture返回一个Observable，但是不会调用你提供的函数，直到有观察者订阅它返回的Observable。这时，它立即调用Future的get()方法，然后镜像发射get()方法返回的Observable发射的数据。

用这种方法，你可以在Observables调用链中包含一个返回Observable的Future对象。

fromAction

fromAction

rxjava-async模块还包含一个fromAction操作符，它接受一个Action作为参数，返回一个Observable，一旦Action终止，它发射这个你传递给fromAction的数据。

fromCallable

fromCallable

rxjava-async模块还包含一个fromCallable操作符，它接受一个Callable作为参数，返回一个发射这个Callable的结果的Observable。

fromRunnable

fromRunnable

rxjava-async模块还包含一个fromRunnable操作符，它接受一个Runnable作为参数，返回一个Observable，一旦Runnable终止，它发射这个你传递给fromRunnable的数据。

forEachFuture

forEachFuture

rxjava-async模块还包含一个forEachFuture操作符。它其实不算Start操作符的一个变体，而是有一些自己的特点。你传递一些典型的观察者方法（如onNext, onError和onCompleted）给它，Observable会以通常的方式调用它。但是forEachFuture自己返回一个Future并且在get()方法处阻塞，直到原始Observable执行完成，然后它返回，完成还是错误依赖于原始Observable是完成还是错误。

如果你想要一个函数阻塞直到Observable执行完成，可以使用这个操作符。

runAsync

rxjava-async模块还包含一个runAsync操作符。它很特殊，返回一个叫做StoppableObservable的特殊Observable。

传递一个Action和一个Scheduler给runAsync，它返回一个使用这个Action产生数据的StoppableObservable。这个Action接受一个Observable和一个Subscription作为参数，它使用Subscription检查unsubscribed条件，一旦发现条件为真就立即停止发射数据。在任何时候你都可以使用unsubscribe方法手动停止一个StoppableObservable（这会同时取消订阅与这个StoppableObservable关联的Subscription）。

由于runAsync会立即调用Action并开始发射数据，在你创建StoppableObservable之后到你的观察者准备好接受数据之前这段时间里，可能会有一部分数据会丢失。如果这不符合你的要求，可以使用runAsync的一个变体，它也接受一个Subject参数，传递一个ReplaySubject给它，你可以获取其它丢失的数据了。

在RxJava中还有一个版本的From操作符可以将Future转换为Observable，与start相似。

Timer

创建一个Observable，它在一个给定的延迟后发射一个特殊的值。

timer

Timer操作符创建一个在给定的时间段之后返回一个特殊值的Observable。

RxJava将这个操作符实现为timer函数。

timer返回一个Observable，它在延迟一段给定的时间后发射一个简单的数字0。

timer操作符默认在computation调度器上执行。有一个变体可以通过可选参数指定Scheduler。

• Javadoc: timer(long,TimeUnit)
• Javadoc: timer(long,TimeUnit,Scheduler)

变换操作

这个页面展示了可用于对Observable发射的数据执行变换操作的各种操作符。

• map( ) — 对序列的每一项都应用一个函数来变换Observable发射的数据序列
• flatMap( ), concatMap( ), and flatMapIterable( ) — 将Observable发射的数据集合变换为Observables集合，然后将这些Observable发射的数据平坦化的放进一个单独的Observable
• switchMap( ) — 将Observable发射的数据集合变换为Observables集合，然后只发射这些Observables最近发射的数据
• scan( ) — 对Observable发射的每一项数据应用一个函数，然后按顺序依次发射每一个值
• groupBy( ) — 将Observable分拆为Observable集合，将原始Observable发射的数据按Key分组，每一个Observable发射一组不同的数据
• buffer( ) — 它定期从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个
• window( ) — 定期将来自Observable的数据分拆成一些Observable窗口，然后发射这些窗口，而不是每次发射一项
• cast( ) — 在发射之前强制将Observable发射的所有数据转换为指定类型

Buffer

定期收集Observable的数据放进一个数据包裹，然后发射这些数据包裹，而不是一次发射一个值。

buffer

Buffer操作符将一个Observable变换为另一个，原来的Observable正常发射数据，变换产生的Observable发射这些数据的缓存集合。Buffer操作符在很多语言特定的实现中有很多种变体，它们在如何缓存这个问题上存在区别。

注意：如果原来的Observable发射了一个onError通知，Buffer会立即传递这个通知，而不是首先发射缓存的数据，即使在这之前缓存中包含了原始Observable发射的数据。

Window操作符与Buffer类似，但是它在发射之前把收集到的数据放进单独的Observable，而不是放进一个数据结构。

在RxJava中有许多Buffer的变体：

buffer(count)

buffer3

buffer(count)以列表(List)的形式发射非重叠的缓存，每一个缓存至多包含来自原始Observable的count项数据（最后发射的列表数据可能少于count项）

• Javadoc: buffer(int)

buffer(count, skip)

buffer4

buffer(count, skip)从原始Observable的第一项数据开始创建新的缓存，此后每当收到skip项数据，用count项数据填充缓存：开头的一项和后续的count-1项，它以列表(List)的形式发射缓存，取决于count和skip的值，这些缓存可能会有重叠部分（比如skip < count时），也可能会有间隙（比如skip > count时）。

• Javadoc: buffer(int,int)

buffer(bufferClosingSelector)

buffer1

当它订阅原来的Observable时，buffer(bufferClosingSelector)开始将数据收集到一个List，然后它调用bufferClosingSelector生成第二个Observable，当第二个Observable发射一个TClosing时，buffer发射当前的List，然后重复这个过程：开始组装一个新的List，然后调用bufferClosingSelector创建一个新的Observable并监视它。它会一直这样做直到原来的Observable执行完成。

• Javadoc: buffer(Func0)

buffer(boundary)

buffer8

buffer(boundary)监视一个名叫boundary的Observable，每当这个Observable发射了一个值，它就创建一个新的List开始收集来自原始Observable的数据并发射原来的List。

• Javadoc: buffer(Observable)
• Javadoc: buffer(Observable,int)

buffer(bufferOpenings, bufferClosingSelector)

buffer2

buffer(bufferOpenings, bufferClosingSelector)监视这个叫bufferOpenings的Observable（它发射BufferOpening对象），每当bufferOpenings发射了一个数据时，它就创建一个新的List开始收集原始Observable的数据，并将bufferOpenings传递给closingSelector函数。这个函数返回一个Observable。buffer监视这个Observable，当它检测到一个来自这个Observable的数据时，就关闭List并且发射它自己的数据（之前的那个List）。

• Javadoc: buffer(Observable,Func1)

buffer(timespan, unit[, scheduler])

buffer5

buffer(timespan, unit)定期以List的形式发射新的数据，每个时间段，收集来自原始Observable的数据（从前面一个数据包裹之后，或者如果是第一个数据包裹，从有观察者订阅原来的Observale之后开始）。还有另一个版本的buffer接受一个Scheduler参数，默认情况下会使用computation调度器。

• Javadoc: buffer(long,TimeUnit)
• Javadoc: buffer(long,TimeUnit,Scheduler)

buffer(timespan, unit, count[, scheduler])

buffer6

每当收到来自原始Observable的count项数据，或者每过了一段指定的时间后，buffer(timespan, unit, count)就以List的形式发射这期间的数据，即使数据项少于count项。还有另一个版本的buffer接受一个Scheduler参数，默认情况下会使用computation调度器。

• Javadoc: buffer(long,TimeUnit,int)
• Javadoc: buffer(long,TimeUnit,int,Scheduler)

buffer(timespan, timeshift, unit[, scheduler])

buffer7

buffer(timespan, timeshift, unit)在每一个timeshift时期内都创建一个新的List,然后用原始Observable发射的每一项数据填充这个列表（在把这个List当做自己的数据发射前，从创建时开始，直到过了timespan这么长的时间）。如果timespan长于timeshift，它发射的数据包将会重叠，因此可能包含重复的数据项。

还有另一个版本的buffer接受一个Scheduler参数，默认情况下会使用computation调度器。

• Javadoc: buffer(long,long,TimeUnit)
• Javadoc: buffer(long,long,TimeUnit,Scheduler)

buffer-backpressure

你可以使用Buffer操作符实现反压backpressure（意思是，处理这样一个Observable：它产生数据的速度可能比它的观察者消费数据的速度快）。

bp.buffer2

Buffer操作符可以将大量的数据序列缩减为较少的数据缓存序列，让它们更容易处理。例如，你可以按固定的时间间隔，定期关闭和发射来自一个爆发性Observable的数据缓存。这相当于一个缓冲区。

示例代码

Observable<List<Integer>> burstyBuffered = bursty.buffer(500, TimeUnit.MILLISECONDS);

bp.buffer1

或者，如果你想更进一步，可以在爆发期将数据收集到缓存，然后在爆发期终止时发射这些数据，使用 Debounce 操作符给buffer操作符发射一个缓存关闭指示器(buffer closing indicator)可以做到这一点。

代码示例：

// we have to multicast the original bursty Observable so we can use it
// both as our source and as the source for our buffer closing selector:
Observable<Integer> burstyMulticast = bursty.publish().refCount();
// burstyDebounced will be our buffer closing selector:
Observable<Integer> burstyDebounced = burstyMulticast.debounce(10, TimeUnit.MILLISECONDS);
// and this, finally, is the Observable of buffers we're interested in:
Observable<List<Integer>> burstyBuffered = burstyMulticast.buffer(burstyDebounced);

参见

FlatMap

FlatMap将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable

flatMap

FlatMap操作符使用一个指定的函数对原始Observable发射的每一项数据执行变换操作，这个函数返回一个本身也发射数据的Observable，然后FlatMap合并这些Observables发射的数据，最后将合并后的结果当做它自己的数据序列发射。

这个方法是很有用的，例如，当你有一个这样的Observable：它发射一个数据序列，这些数据本身包含Observable成员或者可以变换为Observable，因此你可以创建一个新的Observable发射这些次级Observable发射的数据的完整集合。

注意：FlatMap对这些Observables发射的数据做的是合并(merge)操作，因此它们可能是交错的。

在许多语言特定的实现中，还有一个操作符不会让变换后的Observables发射的数据交错，它按照严格的顺序发射这些数据，这个操作符通常被叫作ConcatMap或者类似的名字。

mergeMap

RxJava将这个操作符实现为flatMap函数。

注意：如果任何一个通过这个flatMap操作产生的单独的Observable调用onError异常终止了，这个Observable自身会立即调用onError并终止。

这个操作符有一个接受额外的int参数的一个变体。这个参数设置flatMap从原来的Observable映射Observables的最大同时订阅数。当达到这个限制时，它会等待其中一个终止然后再订阅另一个。

• Javadoc: flatMap(Func1)
• Javadoc: flatMap(Func1,int)

mergeMap.nce

还有一个版本的flatMap为原始Observable的每一项数据和每一个通知创建一个新的Observable（并对数据平坦化）。

它也有一个接受额外int参数的变体。

• Javadoc: flatMap(Func1,Func1,Func0)
• Javadoc: flatMap(Func1,Func1,Func0,int)

mergeMap.r

还有一个版本的flatMap会使用原始Observable的数据触发的Observable组合这些数据，然后发射这些数据组合。它也有一个接受额外int参数的版本。

• Javadoc: flatMap(Func1,Func2)
• Javadoc: flatMap(Func1,Func2,int)

flatMapIterable

mergeMapIterable

flatMapIterable这个变体成对的打包数据，然后生成Iterable而不是原始数据和生成的Observables，但是处理方式是相同的。

• Javadoc: flatMapIterable(Func1)
• Javadoc: flatMapIterable(Func1,Func2)

concatMap

concatMap

还有一个concatMap操作符，它类似于最简单版本的flatMap，但是它按次序连接而不是合并那些生成的Observables，然后产生自己的数据序列。

• Javadoc: concatMap(Func1)

switchMap

switchMap

RxJava还实现了switchMap操作符。它和flatMap很像，除了一点：当原始Observable发射一个新的数据（Observable）时，它将取消订阅并停止监视产生执之前那个数据的Observable，只监视当前这一个。

• Javadoc: switchMap(Func1)

split

St.split

在特殊的StringObservable类（默认没有包含在RxJava中）中还有一个split操作符。它将一个发射字符串的Observable转换为另一个发射字符串的Observable，只不过，后者将原始的数据序列当做一个数据流，使用一个正则表达式边界分割它们，然后合并发射分割的结果。

GroupBy

将一个Observable分拆为一些Observables集合，它们中的每一个发射原始Observable的一个子序列

groupBy

GroupBy操作符将原始Observable分拆为一些Observables集合，它们中的每一个发射原始Observable数据序列的一个子序列。哪个数据项由哪一个Observable发射是由一个函数判定的，这个函数给每一项指定一个Key，Key相同的数据会被同一个Observable发射。

RxJava实现了groupBy操作符。它返回Observable的一个特殊子类GroupedObservable，实现了GroupedObservable接口的对象有一个额外的方法getKey，这个Key用于将数据分组到指定的Observable。

有一个版本的groupBy允许你传递一个变换函数，这样它可以在发射结果GroupedObservable之前改变数据项。

注意：groupBy将原始Observable分解为一个发射多个GroupedObservable的Observable，一旦有订阅，每个GroupedObservable就开始缓存数据。因此，如果你忽略这些GroupedObservable中的任何一个，这个缓存可能形成一个潜在的内存泄露。因此，如果你不想观察，也不要忽略GroupedObservable。你应该使用像take(0)这样会丢弃自己的缓存的操作符。

如果你取消订阅一个GroupedObservable，那个Observable将会终止。如果之后原始的Observable又发射了一个与这个Observable的Key匹配的数据，groupBy将会为这个Key创建一个新的GroupedObservable。

groupBy默认不在任何特定的调度器上执行。

• Javadoc: groupBy(Func1)
• Javadoc: groupBy(Func1,Func1)

Map

对Observable发射的每一项数据应用一个函数，执行变换操作

map

Map操作符对原始Observable发射的每一项数据应用一个你选择的函数，然后返回一个发射这些结果的Observable。

RxJava将这个操作符实现为map函数。这个操作符默认不在任何特定的调度器上执行。

• Javadoc: map(Func1)

cast

cast

cast操作符将原始Observable发射的每一项数据都强制转换为一个指定的类型，然后再发射数据，它是map的一个特殊版本。

• Javadoc: cast(Class)

encode

encode

encode在StringObservable类中，不是标准RxJava的一部分，它也是一个特殊的map操作符。encode将一个发射字符串的Observable变换为一个发射字节数组（这个字节数组按照原始字符串中的多字节字符边界划分）的Observable。

byLine

byLine

byLine同样在StringObservable类中，也不是标准RxJava的一部分，它也是一个特殊的map操作符。byLine将一个发射字符串的Observable变换为一个按行发射来自原始Observable的字符串的Observable。

Scan

连续地对数据序列的每一项应用一个函数，然后连续发射结果

scan

Scan操作符对原始Observable发射的第一项数据应用一个函数，然后将那个函数的结果作为自己的第一项数据发射。它将函数的结果同第二项数据一起填充给这个函数来产生它自己的第二项数据。它持续进行这个过程来产生剩余的数据序列。这个操作符在某些情况下被叫做accumulator。

scan

RxJava实现了scan操作符。

示例代码：

Observable.just(1, 2, 3, 4, 5)
    .scan(new Func2<Integer, Integer, Integer>() {
        @Override
        public Integer call(Integer sum, Integer item) {
            return sum + item;
        }
    }).subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }

        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });

输出

Next: 1
Next: 3
Next: 6
Next: 10
Next: 15
Sequence complete.