RxJava----Utility 辅助操做(响应式处理）

时间  2020-01-09
标签 rxjava utility 辅助 响应 处理 栏目 Java 繁體版
原文   原文链接

这个页面列出了不少用于Observable的辅助操做符html

=========================================================java

Materialize/Dematerialize

Materialize将数据项和事件通知都当作数据项发射，Dematerialize恰好相反。react

materialize

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

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

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

dematerialize

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

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

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

Timestamp

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

Timestamp

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

timestamp默认在immediate调度器上执行，可是能够经过参数指定其它的调度器。

Serialize

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

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

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

Replay

保证全部的观察者收到相同的数据序列，即便它们在Observable开始发射数据以后才订阅

replay

可链接的Observable (connectable Observable)与普通的Observable差很少，不过它并不会在被订阅时开始发射数据，而是直到使用了Connect操做符时才会开始。用这种方法，你能够在任什么时候候让一个Observable开始发射数据。

若是在将一个Observable转换为可链接的Observable以前对它使用Replay操做符，产生的这个可链接Observable将老是发射完整的数据序列给任何将来的观察者，即便那些观察者在这个Observable开始给其它观察者发射数据以后才订阅。

replay

RxJava的实现为replay，它有多个接受不一样参数的变体，有的能够指定replay的最大缓存数量，有的还能够指定调度器。

replay

有一种 replay返回一个普通的Observable。它能够接受一个变换函数为参数，这个函数接受原始Observable发射的数据项为参数，返回结果Observable要发射的一项数据。所以，这个操做符实际上是replay变换以后的数据项。

ObserveOn

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

observeOn

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

observeOn

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

SubscribeOn操做符的做用相似，但它是用于指定Observable自己在特定的调度器上执行，它一样会在那个调度器上给观察者发通知。

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

SubscribeOn

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

SubscribeOn

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

ObserveOn操做符的做用相似，可是功能颇有限，它指示Observable在一个指定的调度器上给观察者发通知。

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

Do

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

do

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

RxJava实现了不少Do操做符的变体。

doOnEach

doOnEach

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

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它就会被调用。

doOnUnsubscribe

doOnUnsubscribe

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

doOnCompleted

doOnCompleted

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

doOnError

doOnError

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

doOnTerminate

doOnTerminate

doOnTerminate 操做符注册一个动做，当它产生的Observable终止以前会被调用，不管是正常仍是异常终止。

finallyDo

finallyDo

finallyDo 操做符注册一个动做，当它产生的Observable终止以后会被调用，不管是正常仍是异常终止。

Delay

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

delay

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

RxJava的实现是 delay和delaySubscription。

delay

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

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

delay默认在computation调度器上执行，你能够经过参数指定使用其它的调度器。

delay

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

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

delay

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

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

delay

还有一个操做符delaySubscription让你你能够延迟订阅原始Observable。它结合搜一个定义延时的参数。

delaySubscription默认在computation调度器上执行，你能够经过参数指定使用其它的调度器。

delay

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

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

TimeInterval

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

TimeInterval

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

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

timeInterval默认在immediate调度器上执行，你能够经过传参数修改。

Using

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

using

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

using

using操做符接受三个参数：

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

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

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

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.
  •  

first(Func1)

first

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

firstOrDefault

firstOrDefault

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

firstOrDefault(Func1)

firstOrDefault

传递一个谓词函数给firstOrDefault，而后发射这个函数断定为true的第一项数据，若是没有数据经过了谓词测试就发射一个默认值。

takeFirst

takeFirst

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

single

single

single操做符也与first相似，可是若是原始Observable在完成以前不是正好发射一次数据，它会抛出一个NoSuchElementException。

single(Func1)

single

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

singleOrDefault

single

和firstOrDefault相似，可是若是原始Observable发射超过一个的数据，会以错误通知终止。

singleOrDefault(T,Func1)

single

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

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发射的数据集。

toFuture

toFuture

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

若是你想将发射多个数据项的Observable转换为Future，能够这样用：myObservable.toList().toBlocking().toFuture()。

toIterable

toIterable

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

toList

toList

一般，发射多项数据的Observable会为每一项数据调用onNext方法。你能够用toList操做符改变这个行为，让Observable将多项数据组合成一个List，而后调用一次onNext方法传递整个列表。

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

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

toMap

toMap

toMap收集原始Observable发射的全部数据项到一个Map（默认是HashMap）而后发射这个Map。你能够提供一个用于生成Map的Key的函数，还能够提供一个函数转换数据项到Map存储的值（默认数据项自己就是值）。

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

toMultiMap

toMultiMap

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

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

toSortedList

toSortedList

toSortedList相似于toList，不一样的是，它会对产生的列表排序，默认是天然升序，若是发射的数据项没有实现Comparable接口，会抛出一个异常。然而，你也能够传递一个函数做为用于比较两个数据项，这是toSortedList不会使用Comparable接口。

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

nest

nest

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

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程