RxJava----Utility 辅助操作

墨蓝 2022-07-29 10:38 200阅读 0赞

这个页面列出了很多用于Observable的辅助操作符

*  **`materialize( )`** — 将Observable转换成一个通知列表convert an Observable into a list of Notifications
 *  **`dematerialize( )`** — 将上面的结果逆转回一个Observable
 *  **`timestamp( )`** — 给Observable发射的每个数据项添加一个时间戳
 *  **`serialize( )`** — 强制Observable按次序发射数据并且要求功能是完好的
 *  **`cache( )`** — 记住Observable发射的数据序列并发射相同的数据序列给后续的订阅者
 *  **`observeOn( )`** — 指定观察者观察Observable的调度器
 *  **`subscribeOn( )`** — 指定Observable执行任务的调度器
 *  **`doOnEach( )`** — 注册一个动作，对Observable发射的每个数据项使用
 *  **`doOnCompleted( )`** — 注册一个动作，对正常完成的Observable使用
 *  **`doOnError( )`** — 注册一个动作，对发生错误的Observable使用
 *  **`doOnTerminate( )`** — 注册一个动作，对完成的Observable使用，无论是否发生错误
 *  **`doOnSubscribe( )`** — 注册一个动作，在观察者订阅时使用
 *  **`doOnUnsubscribe( )`** — 注册一个动作，在观察者取消订阅时使用
 *  **`finallyDo( )`** — 注册一个动作，在Observable完成时使用
 *  **`delay( )`** — 延时发射Observable的结果
 *  **`delaySubscription( )`** — 延时处理订阅请求
 *  **`timeInterval( )`** — 定期发射数据
 *  **`using( )`** — 创建一个只在Observable生命周期存在的资源
 *  **`single( )`** — 强制返回单个数据，否则抛出异常
 *  **`singleOrDefault( )`** — 如果Observable完成时返回了单个数据，就返回它，否则返回默认数据
 *  **`toFuture( )`**, **`toIterable( )`**, **`toList( )`** — 将Observable转换为其它对象或数据结构

=========================================================

## Materialize/Dematerialize ##

`Materialize`将数据项和事件通知都当做数据项发射，`Dematerialize`刚好相反。

![materialize][]

一个合法的有限的Obversable将调用它的观察者的`onNext`方法零次或多次，然后调用观察者的`onCompleted`或`onError`正好一次。`Materialize`操作符将这一系列调用，包括原来的`onNext`通知和终止通知`onCompleted`或`onError`都转换为一个Observable发射的数据序列。

RxJava的`materialize`将来自原始Observable的通知转换为`Notification`对象，然后它返回的Observable会发射这些数据。

`materialize`默认不在任何特定的调度器 (`Scheduler`) 上执行。

*  Javadoc: [materialize()][materialize 1]

![dematerialize][]

`Dematerialize`操作符是`Materialize`的逆向过程，它将`Materialize`转换的结果还原成它原本的形式。

`dematerialize`反转这个过程，将原始Observable发射的`Notification`对象还原成Observable的通知。

`dematerialize`默认不在任何特定的调度器 (`Scheduler`) 上执行。

*  Javadoc: [dematerialize()][dematerialize 1]

## Timestamp ##

给Observable发射的数据项附加一个时间戳

![Timestamp][]

RxJava中的实现为`timestamp`，它将一个发射T类型数据的Observable转换为一个发射类型为`Timestamped<T>`的数据的Observable，每一项都包含数据的原始发射时间。

`timestamp`默认在`immediate`调度器上执行，但是可以通过参数指定其它的调度器。

*  Javadoc: [timestamp()][timestamp]
 *  Javadoc: [timestamp(Scheduler)][timestamp_Scheduler]

## Serialize ##

强制一个Observable连续调用并保证行为正确

![serialize.c.png][]

一个Observable可以异步调用它的观察者的方法，可能是从不同的线程调用。这可能会让Observable行为不正确，它可能会在某一个`onNext`调用之前尝试调用`onCompleted`或`onError`方法，或者从两个不同的线程同时调用`onNext`方法。使用`Serialize`操作符，你可以纠正这个Observable的行为，保证它的行为是正确的且是同步的。

RxJava中的实现是`serialize`，它默认不在任何特定的调度器上执行。

*  Javadoc: [serialize()][serialize]

## Replay ##

保证所有的观察者收到相同的数据序列，即使它们在Observable开始发射数据之后才订阅

![replay][]

可连接的Observable (*connectable Observable*)与普通的Observable差不多，不过它并不会在被订阅时开始发射数据，而是直到使用了`Connect`操作符时才会开始。用这种方法，你可以在任何时候让一个Observable开始发射数据。

如果在将一个Observable转换为可连接的Observable之前对它使用`Replay`操作符，产生的这个可连接Observable将总是发射完整的数据序列给任何未来的观察者，即使那些观察者在这个Observable开始给其它观察者发射数据之后才订阅。

![replay][replay 1]

RxJava的实现为`replay`，它有多个接受不同参数的变体，有的可以指定`replay`的最大缓存数量，有的还可以指定调度器。

*  Javadoc: [replay()][replay 2]
 *  Javadoc: [replay(int)][replay_int]
 *  Javadoc: [replay(long,TimeUnit)][replay_long_TimeUnit]
 *  Javadoc: [replay(int,long,TimeUnit)][replay_int_long_TimeUnit]

![replay][replay 3]

有一种 `replay`返回一个普通的Observable。它可以接受一个变换函数为参数，这个函数接受原始Observable发射的数据项为参数，返回结果Observable要发射的一项数据。因此，这个操作符其实是`replay`变换之后的数据项。

*  Javadoc: [replay(Func1)][replay_Func1]
 *  Javadoc: [replay(Func1,int)][replay_Func1_int]
 *  Javadoc: [replay(Func1,long,TimeUnit)][replay_Func1_long_TimeUnit]
 *  Javadoc: [replay(Func1,int,long,TimeUnit)][replay_Func1_int_long_TimeUnit]

## ObserveOn ##

指定一个观察者在哪个调度器上观察这个Observable

![observeOn][]

很多ReactiveX实现都使用调度器 “`Scheduler`”来管理多线程环境中Observable的转场。你可以使用`ObserveOn`操作符指定Observable在一个特定的调度器上发送通知给观察者 (调用观察者的`onNext`, `onCompleted`, `onError`方法)。

![observeOn][observeOn 1]

注意：当遇到一个异常时`ObserveOn`会立即向前传递这个`onError`终止通知，它不会等待慢速消费的Observable接受任何之前它已经收到但还没有发射的数据项。这可能意味着`onError`通知会跳到（并吞掉）原始Observable发射的数据项前面，正如图例上展示的。

`SubscribeOn`操作符的作用类似，但它是用于指定Observable本身在特定的调度器上执行，它同样会在那个调度器上给观察者发通知。

RxJava中，要指定Observable应该在哪个调度器上调用观察者的`onNext`, `onCompleted`, `onError`方法，你需要使用`observeOn`操作符，传递给它一个合适的`Scheduler`。

*  Javadoc: [observeOn(Scheduler)][observeOn_Scheduler]

## SubscribeOn ##

指定Observable自身在哪个调度器上执行

![SubscribeOn][]

很多ReactiveX实现都使用调度器 “`Scheduler`”来管理多线程环境中Observable的转场。你可以使用`SubscribeOn`操作符指定Observable在一个特定的调度器上运转。

`ObserveOn`操作符的作用类似，但是功能很有限，它指示Observable在一个指定的调度器上给观察者发通知。

在某些实现中还有一个`UnsubscribeOn`操作符。

*  Javadoc: [subscribeOn(Scheduler)][subscribeOn_Scheduler]
 *  Javadoc: [unsubscribeOn(Scheduler)][unsubscribeOn_Scheduler]

## Do ##

注册一个动作作为原始Observable生命周期事件的一种占位符

![do][]

你可以注册回调，当Observable的某个事件发生时，Rx会在与Observable链关联的正常通知集合中调用它。Rx实现了多种操作符用于达到这个目的。

RxJava实现了很多`Do`操作符的变体。

### doOnEach ###

![doOnEach][]

`doOnEach`操作符让你可以注册一个回调，它产生的Observable每发射一项数据就会调用它一次。你可以以`Action`的形式传递参数给它，这个Action接受一个`onNext`的变体`Notification`作为它的唯一参数，你也可以传递一个Observable给`doOnEach`，这个Observable的`onNext`会被调用，就好像它订阅了原始的Observable一样。

*  Javadoc: [doOnEach(Action1)][doOnEach_Action1]
 *  Javadoc: [doOnEach(Observer)][doOnEach_Observer]

### doOnNext ###

![doOnNext][]

`doOnNext`操作符类似于`doOnEach(Action1)`，但是它的Action不是接受一个`Notification`参数，而是接受发射的数据项。

示例代码

Observable.just(1, 2, 3)
              .doOnNext(new Action1<Integer>() {
              @Override
              public void call(Integer item) {
                if( item > 1 ) {
                  throw new RuntimeException( "Item exceeds maximum value" );
                }
              }
            }).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
    Error: Item exceeds maximum value

### doOnSubscribe ###

![doOnSubscribe][]

`doOnSubscribe`操作符注册一个动作，当观察者订阅它生成的Observable它就会被调用。

*  Javadoc: [doOnSubscribe(Action0)][doOnSubscribe_Action0]

### doOnUnsubscribe ###

![doOnUnsubscribe][]

`doOnUnsubscribe`操作符注册一个动作，当观察者取消订阅它生成的Observable它就会被调用。

*  Javadoc: [doOnUnsubscribe(Action0)][doOnUnsubscribe_Action0]

### doOnCompleted ###

![doOnCompleted][]

`doOnCompleted` 操作符注册一个动作，当它产生的Observable正常终止调用`onCompleted`时会被调用。

*  Javadoc: [doOnCompleted(Action0)][doOnCompleted_Action0]

### doOnError ###

![doOnError][]

`doOnError` 操作符注册一个动作，当它产生的Observable异常终止调用`onError`时会被调用。

*  Javadoc: [doOnError(Action0)][doOnError_Action0]

### doOnTerminate ###

![doOnTerminate][]

`doOnTerminate` 操作符注册一个动作，当它产生的Observable终止之前会被调用，无论是正常还是异常终止。

*  Javadoc: [doOnTerminate(Action0)][doOnTerminate_Action0]

### finallyDo ###

![finallyDo][]

`finallyDo` 操作符注册一个动作，当它产生的Observable终止之后会被调用，无论是正常还是异常终止。

*  Javadoc: [finallyDo(Action0)][finallyDo_Action0]

# Delay #

延迟一段指定的时间再发射来自Observable的发射物

![delay][]

`Delay`操作符让原始Observable在发射每项数据之前都暂停一段指定的时间段。效果是Observable发射的数据项在时间上向前整体平移了一个增量。

RxJava的实现是 `delay`和`delaySubscription`。

![delay][delay 1]

第一种`delay`接受一个定义时长的参数（包括数量和单位）。每当原始Observable发射一项数据，`delay`就启动一个定时器，当定时器过了给定的时间段时，`delay`返回的Observable发射相同的数据项。

注意：`delay`不会平移`onError`通知，它会立即将这个通知传递给订阅者，同时丢弃任何待发射的`onNext`通知。然而它会平移一个`onCompleted`通知。

`delay`默认在`computation`调度器上执行，你可以通过参数指定使用其它的调度器。

*  Javadoc: [delay(long,TimeUnit)][delay_long_TimeUnit]
 *  Javadoc: [delay()][delay 2]

![delay][delay 3]

另一种`delay`不实用常数延时参数，它使用一个函数针对原始Observable的每一项数据返回一个Observable，它监视返回的这个Observable，当任何那样的Observable终止时，`delay`返回的Observable就发射关联的那项数据。

这种`delay`默认不在任何特定的调度器上执行。

*  Javadoc: [delay(Func1)][delay_Func1]

![delay][delay 4]

这个版本的`delay`对每一项数据使用一个Observable作为原始Observable的延时定时器。

这种`delay`默认不在任何特定的调度器上执行。

*  Javadoc: [delay(Func0,Func1)][delay_Func0_Func1]

![delay][delay 5]

还有一个操作符`delaySubscription`让你你可以延迟订阅原始Observable。它结合搜一个定义延时的参数。

`delaySubscription`默认在`computation`调度器上执行，你可以通过参数指定使用其它的调度器。

*  Javadoc: [delaySubscription(long,TimeUnit)][delaySubscription_long_TimeUnit]
 *  Javadoc: [delaySubscription(long,TimeUnit,Scheduler)][delaySubscription_long_TimeUnit_Scheduler]

![delay][delay 6]

还有一个版本的`delaySubscription`使用一个Obseable而不是一个固定的时长来设置订阅延时。

这种`delaySubscription`默认不在任何特定的调度器上执行。

*  Javadoc: [delaySubscription(Func0)][delaySubscription_Func0]

## TimeInterval ##

将一个发射数据的Observable转换为发射那些数据发射时间间隔的Observable

![TimeInterval][]

`TimeInterval`操作符拦截原始Observable发射的数据项，替换为发射表示相邻发射物时间间隔的对象。

RxJava中的实现为`timeInterval`，这个操作符将原始Observable转换为另一个Obserervable，后者发射一个标志替换前者的数据项，这个标志表示前者的两个连续发射物之间流逝的时间长度。新的Observable的第一个发射物表示的是在观察者订阅原始Observable到原始Observable发射它的第一项数据之间流逝的时间长度。不存在与原始Observable发射最后一项数据和发射`onCompleted`通知之间时长对应的发射物。

`timeInterval`默认在`immediate`调度器上执行，你可以通过传参数修改。

*  Javadoc: [timeInterval()][timeInterval]
 *  Javadoc: [timeInterval(Scheduler)][timeInterval_Scheduler]

## Using ##

创建一个只在Observable生命周期内存在的一次性资源

![using][]

`Using`操作符让你可以指示Observable创建一个只在它的生命周期内存在的资源，当Observable终止时这个资源会被自动释放。

![using][]

`using`操作符接受三个参数：

1.  一个用户创建一次性资源的工厂函数
2.  一个用于创建Observable的工厂函数
3.  一个用于释放资源的函数

当一个观察者订阅`using`返回的Observable时，`using`将会使用Observable工厂函数创建观察者要观察的Observable，同时使用资源工厂函数创建一个你想要创建的资源。当观察者取消订阅这个Observable时，或者当观察者终止时（无论是正常终止还是因错误而终止），`using`使用第三个函数释放它创建的资源。

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

*  Javadoc: [using(Func0,Func1,Action1)][using_Func0_Func1_Action1]

## First ##

只发射第一项（或者满足某个条件的第一项）数据

![first][]

如果你只对Observable发射的第一项数据，或者满足某个条件的第一项数据感兴趣，你可以使用`First`操作符。

在某些实现中，`First`没有实现为一个返回Observable的过滤操作符，而是实现为一个在当时就发射原始Observable指定数据项的阻塞函数。在这些实现中，如果你想要的是一个过滤操作符，最好使用`Take(1)`或者`ElementAt(0)`。

在一些实现中还有一个`Single`操作符。它的行为与`First`类似，但为了确保只发射单个值，它会等待原始Observable终止（否则，不是发射那个值，而是以一个错误通知终止）。你可以使用它从原始Observable获取第一项数据，而且也确保只发射一项数据。

在RxJava中，这个操作符被实现为`first`，`firstOrDefault`和`takeFirst`。

可能容易混淆，`BlockingObservable`也有名叫`first`和`firstOrDefault`的操作符，它们会阻塞并返回值，不是立即返回一个Observable。

还有几个其它的操作符执行类似的功能。

### 过滤操作符 ###

![first][]

只发射第一个数据，使用没有参数的`first`操作符。

示例代码

Observable.just(1, 2, 3)
              .first()
              .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
    Sequence complete.

*  Javadoc: [first()][first 1]

### first(Func1) ###

![first][first 2]

传递一个谓词函数给`first`，然后发射这个函数判定为`true`的第一项数据。

*  Javadoc: [first(Func1)][first_Func1]

### firstOrDefault ###

![firstOrDefault][]

`firstOrDefault`与`first`类似，但是在Observagle没有发射任何数据时发射一个你在参数中指定的默认值。

*  Javadoc: [firstOrDefault(T)][firstOrDefault_T]

### firstOrDefault(Func1) ###

![firstOrDefault][firstOrDefault 1]

传递一个谓词函数给`firstOrDefault`，然后发射这个函数判定为`true`的第一项数据，如果没有数据通过了谓词测试就发射一个默认值。

*  Javadoc [firstOrDefault(T, Func1)][firstOrDefault_T_ Func1]

### takeFirst ###

![takeFirst][]

`takeFirst`与`first`类似，除了这一点：如果原始Observable没有发射任何满足条件的数据，`first`会抛出一个`NoSuchElementException`，`takeFist`会返回一个空的Observable（不调用`onNext()`但是会调用`onCompleted`）。

*  Javadoc: [takeFirst(Func1)][takeFirst_Func1]

### single ###

![single][]

`single`操作符也与`first`类似，但是如果原始Observable在完成之前不是正好发射一次数据，它会抛出一个`NoSuchElementException`。

*  Javadoc: [single()][single 1]

### single(Func1) ###

![single][single 2]

`single`的变体接受一个谓词函数，发射满足条件的单个值，如果不是正好只有一个数据项满足条件，会以错误通知终止。

*  Javadoc: [single(Func1)][single_Func1]

### singleOrDefault ###

![single][single 3]

和`firstOrDefault`类似，但是如果原始Observable发射超过一个的数据，会以错误通知终止。

*  Javadoc: [singleOrDefault(T)][singleOrDefault_T]

### singleOrDefault(T,Func1) ###

![single][single 4]

和`firstOrDefault(T, Func1)`类似，如果没有数据满足条件，返回默认值；如果有多个数据满足条件，以错误通知终止。

*  Javadoc: [singleOrDefault(Func1,T)][singleOrDefault_Func1_T]

first系列的这几个操作符默认不在任何特定的调度器上执行。

## To ##

将Observable转换为另一个对象或数据结构

![to][]

ReactiveX的很多语言特定实现都有一种操作符让你可以将Observable或者Observable发射的数据序列转换为另一个对象或数据结构。它们中的一些会阻塞直到Observable终止，然后生成一个等价的对象或数据结构；另一些返回一个发射那个对象或数据结构的Observable。

在某些ReactiveX实现中，还有一个操作符用于将Observable转换成阻塞式的。一个阻塞式的Ogbservable在普通的Observable的基础上增加了几个方法，用于操作Observable发射的数据项。

### getIterator ###

![getIterator][]

`getIterator`操作符只能用于`BlockingObservable`的子类，要使用它，你首先必须把原始的Observable转换为一个`BlockingObservable`。可以使用这两个操作符：`BlockingObservable.from`或`the Observable.toBlocking`。

这个操作符将Observable转换为一个`Iterator`，你可以通过它迭代原始Observable发射的数据集。

*  Javadoc: [BlockingObservable.getIterator()][BlockingObservable.getIterator]

### toFuture ###

![toFuture][]

`toFuture`操作符也是只能用于`BlockingObservable`。这个操作符将Observable转换为一个返回单个数据项的`Future`，如果原始Observable发射多个数据项，`Future`会收到一个`IllegalArgumentException`；如果原始Observable没有发射任何数据，`Future`会收到一个`NoSuchElementException`。

如果你想将发射多个数据项的Observable转换为`Future`，可以这样用：`myObservable.toList().toBlocking().toFuture()`。

*  Javadoc: [BlockingObservable.toFuture()][BlockingObservable.toFuture]

### toIterable ###

![toIterable][]

`toFuture`操作符也是只能用于`BlockingObservable`。这个操作符将Observable转换为一个`Iterable`，你可以通过它迭代原始Observable发射的数据集。

*  Javadoc: [BlockingObservable.toIterable()][BlockingObservable.toIterable]

### toList ###

![toList][]

通常，发射多项数据的Observable会为每一项数据调用`onNext`方法。你可以用`toList`操作符改变这个行为，让Observable将多项数据组合成一个`List`，然后调用一次`onNext`方法传递整个列表。

如果原始Observable没有发射任何数据就调用了`onCompleted`，`toList`返回的Observable会在调用`onCompleted`之前发射一个空列表。如果原始Observable调用了`onError`，`toList`返回的Observable会立即调用它的观察者的`onError`方法。

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

*  Javadoc: [toList()][toList 1]

### toMap ###

![toMap][]

`toMap`收集原始Observable发射的所有数据项到一个Map（默认是HashMap）然后发射这个Map。你可以提供一个用于生成Map的Key的函数，还可以提供一个函数转换数据项到Map存储的值（默认数据项本身就是值）。

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

*  Javadoc: [toMap(Func1)][toMap_Func1]
 *  Javadoc: [toMap(Func1,Func1)][toMap_Func1_Func1]
 *  Javadoc: [toMap(Func1,Func1,Func0)][toMap_Func1_Func1_Func0]

### toMultiMap ###

![toMultiMap][]

`toMultiMap`类似于`toMap`，不同的是，它生成的这个Map同时还是一个`ArrayList`（默认是这样，你可以传递一个可选的工厂方法修改这个行为）。

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

*  Javadoc: [toMultiMap(Func1)][toMultiMap_Func1]
 *  Javadoc: [toMultiMap(Func1,Func1)][toMultiMap_Func1_Func1]
 *  Javadoc: [toMultiMap(Func1,Func1,Func0)][toMultiMap_Func1_Func1_Func0]
 *  Javadoc: [toMultiMap(Func1,Func1,Func0,Func1)][toMultiMap_Func1_Func1_Func0_Func1]

### toSortedList ###

![toSortedList][]

`toSortedList`类似于`toList`，不同的是，它会对产生的列表排序，默认是自然升序，如果发射的数据项没有实现`Comparable`接口，会抛出一个异常。然而，你也可以传递一个函数作为用于比较两个数据项，这是`toSortedList`不会使用`Comparable`接口。

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

*  Javadoc: [toSortedList()][toSortedList 1]
 *  Javadoc: [toSortedList(Func2)][toSortedList_Func2]

### nest ###

![nest][]

`nest`操作符有一个特殊的用途：将一个Observable转换为一个发射这个Observable的Observable。

[materialize]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/materialize.c.png
[materialize 1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#materialize%28%29
[dematerialize]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/dematerialize.c.png
[dematerialize 1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#dematerialize%28%29
[Timestamp]: /images/20220728/c48de6e6a74947cea2ec97cf3bbc25cc.png
[timestamp]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#timestamp%28%29
[timestamp_Scheduler]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#timestamp%28rx.Scheduler%29
[serialize.c.png]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/serialize.c.png
[serialize]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#serialize%28%29
[replay]: https://github.com/mcxiaoke/RxDocs/blob/master/images/operators/replay.c.png
[replay 1]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/replay.png
[replay 2]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#replay%28%29
[replay_int]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#replay%28int%29
[replay_long_TimeUnit]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#replay%28long,%20java.util.concurrent.TimeUnit%29
[replay_int_long_TimeUnit]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#replay%28int,%20long,%20java.util.concurrent.TimeUnit%29
[replay 3]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/replay.f.png
[replay_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#replay%28rx.functions.Func1%29
[replay_Func1_int]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#replay%28rx.functions.Func1,%20int%29
[replay_Func1_long_TimeUnit]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#replay%28rx.functions.Func1,%20long,%20java.util.concurrent.TimeUnit%29
[replay_Func1_int_long_TimeUnit]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#replay%28rx.functions.Func1,%20int,%20long,%20java.util.concurrent.TimeUnit%29
[observeOn]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/observeOn.c.png
[observeOn 1]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/observeOn.e.png
[observeOn_Scheduler]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#observeOn%28rx.Scheduler%29
[SubscribeOn]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/subscribeOn.c.png
[subscribeOn_Scheduler]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#subscribeOn%28rx.Scheduler%29
[unsubscribeOn_Scheduler]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#unsubscribeOn%28rx.Scheduler%29
[do]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/do.c.png
[doOnEach]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/doOnEach.png
[doOnEach_Action1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#doOnEach%28rx.functions.Action1%29
[doOnEach_Observer]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#doOnEach%28rx.Observer%29
[doOnNext]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/doOnNext.png
[doOnSubscribe]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/doOnSubscribe.png
[doOnSubscribe_Action0]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#doOnSubscribe%28rx.functions.Action0%29
[doOnUnsubscribe]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/doOnUnsubscribe.png
[doOnUnsubscribe_Action0]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#doOnUnsubscribe%28rx.functions.Action0%29
[doOnCompleted]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/doOnCompleted.png
[doOnCompleted_Action0]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#doOnCompleted%28rx.functions.Action0%29
[doOnError]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/doOnError.png
[doOnError_Action0]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#doOnError%28rx.functions.Action1%29
[doOnTerminate]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/doOnTerminate.png
[doOnTerminate_Action0]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#doOnTerminate%28rx.functions.Action0%29
[finallyDo]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/finallyDo.png
[finallyDo_Action0]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#finallyDo%28rx.functions.Action0%29
[delay]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/delay.c.png
[delay 1]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/delay.png
[delay_long_TimeUnit]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#delay%28long,%20java.util.concurrent.TimeUnit%29
[delay 2]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#delay%28long,%20java.util.concurrent.TimeUnit,%20rx.Scheduler%29
[delay 3]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/delay.o.png
[delay_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#delay%28rx.functions.Func1%29
[delay 4]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/delay.oo.png
[delay_Func0_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#delay%28rx.functions.Func0,%20rx.functions.Func1%29
[delay 5]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/delaySubscription.png
[delaySubscription_long_TimeUnit]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#delaySubscription%28long,%20java.util.concurrent.TimeUnit%29
[delaySubscription_long_TimeUnit_Scheduler]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#delaySubscription%28long,%20java.util.concurrent.TimeUnit,%20rx.Scheduler%29
[delay 6]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/delaySubscription.o.png
[delaySubscription_Func0]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#delaySubscription%28rx.functions.Func0%29
[TimeInterval]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/timeInterval.c.png
[timeInterval]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#timeInterval%28%29
[timeInterval_Scheduler]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#timeInterval%28rx.Scheduler%29
[using]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/using.png
[using_Func0_Func1_Action1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#using%28rx.functions.Func0,%20rx.functions.Func1,%20rx.functions.Action1%29
[first]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/first.png
[first 1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#first%28%29
[first 2]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/firstN.png
[first_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#first%28rx.functions.Func1%29
[firstOrDefault]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/firstOrDefault.png
[firstOrDefault_T]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#firstOrDefault%28T%29
[firstOrDefault 1]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/firstOrDefaultN.png
[firstOrDefault_T_ Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#firstOrDefault%28T,%20rx.functions.Func1%29
[takeFirst]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/first.takeFirst.png
[takeFirst_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#takeFirst%28rx.functions.Func1%29
[single]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/single.png
[single 1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#single%28%29
[single 2]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/single.p.png
[single_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#single%28rx.functions.Func1%29
[single 3]: /images/20220728/7075559349e3402d8bfb1fbfa83b99de.png
[singleOrDefault_T]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#singleOrDefault%28T%29
[single 4]: /images/20220728/af2fafdf95cc4af68961d246c533e6c1.png
[singleOrDefault_Func1_T]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#singleOrDefault%28rx.functions.Func1,%20T%29
[to]: /images/20220728/5c945a16c6314dc6b0cb44bc585b5ca5.png
[getIterator]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/B.getIterator.png
[BlockingObservable.getIterator]: http://reactivex.io/RxJava/javadoc/rx/observables/BlockingObservable.html#getIterator%28%29
[toFuture]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/B.toFuture.png
[BlockingObservable.toFuture]: http://reactivex.io/RxJava/javadoc/rx/observables/BlockingObservable.html#toFuture%28%29
[toIterable]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/B.toIterable.png
[BlockingObservable.toIterable]: http://reactivex.io/RxJava/javadoc/rx/observables/BlockingObservable.html#toIterable%28%29
[toList]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/toList.png
[toList 1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#toList%28%29
[toMap]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/toMap.png
[toMap_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#toMap%28rx.functions.Func1%29
[toMap_Func1_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#toMap%28rx.functions.Func1,%20rx.functions.Func1%29
[toMap_Func1_Func1_Func0]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#toMap%28rx.functions.Func1,%20rx.functions.Func1,%20rx.functions.Func0%29
[toMultiMap]: /images/20220728/94f2f4f68cee4a19969e2414e7ad2dff.png
[toMultiMap_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#toMultimap%28rx.functions.Func1%29
[toMultiMap_Func1_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#toMultimap%28rx.functions.Func1,%20rx.functions.Func1%29
[toMultiMap_Func1_Func1_Func0]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#toMultimap%28rx.functions.Func1,%20rx.functions.Func1,%20rx.functions.Func0%29
[toMultiMap_Func1_Func1_Func0_Func1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#toMultimap%28rx.functions.Func1,%20rx.functions.Func1,%20rx.functions.Func0,%20rx.functions.Func1%29
[toSortedList]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/toSortedList.png
[toSortedList 1]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#toSortedList%28%29
[toSortedList_Func2]: http://reactivex.io/RxJava/javadoc/rx/Observable.html#toSortedList%28rx.functions.Func2%29
[nest]: https://github.com/mcxiaoke/RxDocs/raw/master/images/operators/nest.png