Observable

在ReactiveX中，观察者订阅了一个Observable。然后，该观察者对Observable发出的任何项目或项目序列做出反应。

这种模式有利于并发操作，因为它不需要在等待Observable发出对象时阻塞，而是以观察者的形式创建一个哨兵，随时准备在Observable所做的任何时候做出适当的反应。这个页面解释了反应模式是什么以及Observables和观察者是什么（以及观察者如何订阅Observables）。

其他页面显示了如何使用各种Observable运算符将Observable链接在一起并更改其行为。本文档附有“大理石图表”的说明。以下是大理石图表如何表示Observables的Observables和转换：

Background

在许多软件编程任务中，您或多或少地期望您编写的指令将按照您编写的顺序逐个执行并逐步完成。但是在ReactiveX中，许多指令可以并行执行，其结果随后由“观察者”按任意顺序捕获。而不是调用方法，而是以“Observable”的形式定义检索和转换数据的机制。，“然后订阅一个观察者，此时，先前定义的机制开始行动，观察员站立哨兵，以便在准备好时捕获并响应其排放。

这种方法的一个优点是，当你有一堆不依赖于彼此的任务时，你可以同时启动所有任务，而不是等到每个任务完成后再开始下一个 - 这样，你的整个捆绑任务只需要与捆绑中最长的任务一样长。

有许多术语用于描述这种异步编程和设计模型。本文档将使用以下术语：观察者订阅Observable。Observable通过调用观察者的方法发出项目或向其观察者发送通知。

在其他文件和其他背景下，我们所谓的“观察者”有时被称为“用户”，“观察者”或“反应堆”。这种模型通常被称为“反应堆模式”。

Establishing Observers

此页面使用类似Groovy的伪代码作为示例，但在许多语言中都有ReactiveX实现。

在普通的方法调用中 - 也就是说，不是ReactiveX中典型的异步并行调用 - 流程是这样的：

调用方法。
将该方法的返回值存储在变量中。
使用该变量及其新值来做一些有用的事情。

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

在异步模型中，流程更像是：

定义一个方法，该方法对异步调用的返回值执行一些有用的操作;这种方法是观察者的一部分。
将异步调用本身定义为Observable。
通过订阅将观察者附加到该Observable（这也启动了Observable的操作）。
继续你的事业;每当调用返回时，观察者的方法将开始对其返回值或值进行操作 - 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, and onError

Subscribe方法是将观察者连接到Observable的方法。您的观察者实现以下方法的某些子集：

onNext

只要Observable发出一个项目，Observable就会调用此方法。此方法将Observable发出的项作为参数。

onError

Observable调用此方法以指示它无法生成预期数据或遇到其他一些错误。它不会进一步调用onNext或onCompleted。onError方法将参数指示导致错误的原因。

onCompleted

如果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也有可能产生并尝试发射物品一段时间。

Some Notes on Naming Conventions

ReactiveX的每种语言特定实现都有自己的命名怪癖。虽然实现之间存在许多共性，但没有规范的命名标准。

此外，这些名称中的一些在其他情境中具有不同的含义，或者在特定实现语言的习语中看起来很尴尬。例如，有onEvent命名模式（例如onNext，onCompleted，onError）。

在某些上下文中，这些名称将指示通过哪些方法注册事件处理程序。但是，在ReactiveX中，它们自己命名事件处理程序。

“Hot” and “Cold” Observables

Observable何时开始发出其物品序列？这取决于Observable。“热”Observable可以在创建项目后立即开始发出项目，因此任何后来订阅该Observable的观察者都可以开始在中间某处观察序列。另一方面，“冷”Observable等待观察者在开始发射物品之前订阅它，因此这样的观察者保证从一开始就看到整个序列。

在ReactiveX的一些实现中，还存在称为“可连接”可观察的东西。这样的Observable在调用Connect方法之前不会开始发出项目，无论是否有任何观察者都订阅了它。

Composition via Observable Operators

Observable和观察者只是ReactiveX的开始。

它们本身只不过是标准观察者模式的轻微扩展，更适合处理一系列事件而不是单个回调。

真正的力量来自“反应式扩展”（因此是“ReactiveX”） - 允许您转换，组合，操作和处理Observable发出的项目序列的运算符。

这些Rx运算符允许您以声明方式组合异步序列，同时具有回调的所有效率优势，但没有嵌套通常与异步系统相关联的回调处理程序的缺点。

本文档将有关各种运算符的信息及其用法示例分组到以下页面中：

Creating Observables

Create, Defer, Empty/Never/Throw, From, Interval, Just, Range, Repeat, Start, and Timer

Transforming Observable Items

Buffer, FlatMap, GroupBy, Map, Scan, and Window

Filtering Observables

Debounce, Distinct, ElementAt, Filter, First, IgnoreElements, Last, Sample, Skip, SkipLast, Take, and TakeLast

Combining Observables

And/Then/When, CombineLatest, Join, Merge, StartWith, Switch, and Zip

Error Handling Operators

Catch and Retry

Utility Operators

Delay, Do, Materialize/Dematerialize, ObserveOn, Serialize, Subscribe, SubscribeOn,TimeInterval, Timeout, Timestamp, and Using