RxJava入门实践④

最近大家都在学习RXJAVA，作为有理想(mei shi gan)的程序员，RxJava还是沈有必要学习下的。

RXJAVA地址： https://github.com/ReactiveX/RxJava

0.什么是观察者模式

1.RXJAVA是什么

在Rxjava当中最重要的就是操作符，RxJava当中有着庞大的操作符

创建操作符：负责创建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

变换操作符：对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发射的所有数据转换为指定类型

过滤操作符：用于过滤和选择Observable发射的数据序列

• filter( ) — 过滤数据
• takeLast( ) — 只发射最后的N项数据
• last( ) — 只发射最后的一项数据
• lastOrDefault( ) — 只发射最后的一项数据，如果Observable为空就发射默认值
• takeLastBuffer( ) — 将最后的N项数据当做单个数据发射
• skip( ) — 跳过开始的N项数据
• skipLast( ) — 跳过最后的N项数据
• take( ) — 只发射开始的N项数据
• first( ) and takeFirst( ) — 只发射第一项数据，或者满足某种条件的第一项数据
• firstOrDefault( ) — 只发射第一项数据，如果Observable为空就发射默认值
• elementAt( ) — 发射第N项数据
• elementAtOrDefault( ) — 发射第N项数据，如果Observable数据少于N项就发射默认值
• sample( ) or throttleLast( ) — 定期发射Observable最近的数据
• throttleFirst( ) — 定期发射Observable发射的第一项数据
• throttleWithTimeout( ) or debounce( ) — 只有当Observable在指定的时间后还没有发射数据时，才发射一个数据
• timeout( ) — 如果在一个指定的时间段后还没发射数据，就发射一个异常
• distinct( ) — 过滤掉重复数据
• distinctUntilChanged( ) — 过滤掉连续重复的数据
• ofType( ) — 只发射指定类型的数据
• ignoreElements( ) — 丢弃所有的正常数据，只发射错误或完成通知

结合操作符：用于组合多个Observables的组合使用

• startWith( ) — 在数据序列的开头增加一项数据
• merge( ) — 将多个Observable合并为一个
• mergeDelayError( ) — 合并多个Observables，让没有错误的Observable都完成后再发射错误通知
• zip( ) — 使用一个函数组合多个Observable发射的数据集合，然后再发射这个结果
• and( ), then( ), and when( ) — (rxjava-joins) 通过模式和计划组合多个Observables发射的数据集合
• combineLatest( ) — 当两个Observables中的任何一个发射了一个数据时，通过一个指定的函数组合每个Observable发射的最新数据（一共两个数据），然后发射这个函数的结果
• join( ) and groupJoin( ) — 无论何时，如果一个Observable发射了一个数据项，只要在另一个Observable发射的数据项定义的时间窗口内，就将两个Observable发射的数据合并发射
• switchOnNext( ) — 将一个发射Observables的Observable转换成另一个Observable，后者发射这些Observables最近发射的数据

辅助操作符：用于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转换为其它对象或数据结构

还有一些 字符串处理、连接操作、算数据和操作等不再赘述。

客官息怒，这么多的操作符让大家短时间内接受当然是不现实的，我们主要就是根据操作符的分类来学习一定类别的操作符是怎么使用的就可以了，需要使用的时候只需要到相应分类下去找到特定的操作符就好了。

以创建操作符为例：

在前面的例子中我们学习了如何使用最简单的方式create来创建Observable对象

我们想要使用以前的例子依次打印 “hello”、”RXJAVA“、“this”、“is”、”a“、”demo“这几个字符串我们该如何操作呢？

显然我们应该这么做：

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {     @Override     public void call(Subscriber<? super String> subscriber) {         subscriber.onNext("hello");         subscriber.onNext("RxJava");         subscriber.onNext("this");         subscriber.onNext("is");         subscriber.onNext("a");         subscriber.onNext("demo");         subscriber.onCompleted();     }  });

但是这样的操作真的有必要吗？RxJAVA中可以使用jsut和from创建操作符轻松完成上面的功能。

使用Just：

Observable observable=Observable .just("hello","RxJava","this","is","a","demo");

使用from：

String infos[]={"hello","RxJava","this","is","a","demo"}; Observable observable=Observable.from(infos);

效果如下：

以变换操作符为例：

假设有这么一个场景根据若干数字数组求前一个递归和序列并转换为string字符数组输出。

如：1、2、3、4、5

输出：[“1″,“3”,”6″,”10″,”15”]

Integer numbers[]={1,2,3,4,5,}; Observable observable=Observable.from(numbers); observable         .scan(new Func2<Integer,Integer,Integer>() {             @Override             public Integer call(Integer sum, Integer item) {                 return sum+item;             }         })         .map(new Func1<Integer,String>() {             @Override             public String call(Integer number) {                 return String.valueOf(number);             }         })         .subscribe(subscriber);

首先我们使用From操作符创建Observable然后使用scan操作符来进行数据变换操作，然后我们使用map来进行数据格式变换操作来讲int类型的数据转换为String类型的数据。

scan操作符：

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

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

Map操作符：

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

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

Func，是RXJAVA中自带接口函数（有返回值），从Func1到func9可以传入多个参数，比如func1（String，String）第一个参数为传入的参数类型第二个参数为返回值的类型。

Action1-Action9和Func类似，只不过Action是没有返回值的。

以过滤操作为例：

情景：假设我们想要得到上述计算结果为偶数的字符数组呢？

在scan后增加过滤操作就ok了

.filter(new Func1<Integer,Boolean>() {     @Override     public Boolean call(Integer number) {             return (number%2==0);     } })

那么符合条件的就只有6和10了

filter：发射通过过滤的数据，在func1中第一个参数为参数类型，第二个参数为boolean类型为返回值类型，返回true代表通过过滤，返回false代表过滤失败，不会进行发射。

以结合操作符为例：

结合操作符可以让Observables组合使用，下面以zip为例。

有三个不同的Observable为别获取一个对象的size、color、shape，三者的数据类型可以任意定义。

我们可以使用zip操作符把三者获取的属性整合为具有三者属性的一个对象。

Zip

通过一个函数将多个Observables的发射物结合到一起，基于这个函数的结果为每个结合体发射单个数据项。

Zip操作符返回一个Obversable，它使用这个函数按顺序结合两个或多个Observables发射的数据项，然后它发射这个函数返回的结果。它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个Observable一样多的数据。

RxJava将这个操作符实现为zip和zipWith。

zip的最后一个参数接受每个Observable发射的一项数据，返回被压缩后的数据，它可以接受一到九个参数：一个Observable序列，或者一些发射Observable的Observables。

以辅助操作符为例：

在前面我们使用的线程调度例子中

observeOn( ) — 指定观察者观察Observable的调度器subscribeOn( ) — 指定Observable执行任务的调度器

都属于辅助操作符，当然操作符有很多如：

delay( ) — 延时发射Observable的结果delaySubscription( ) — 延时处理订阅请求

我们可以在以前的例子中使用这个操作符试验下。

.delay(2,TimeUnit.SECONDS)//延迟两秒发送 .delaySubscription(2,TimeUnit.SECONDS)//延迟两秒订阅 .subscribeOn(Schedulers.io())//调度器修改执行线程 .observeOn(AndroidSchedulers.mainThread())//修改执行线程

经过这样的操作你就会发现结果会在4秒后开始显示

当然RxJava的操作符有很多，我们不可能都掌握它，但是知道那些操作符属于哪一类是很关键的，因为只有这样你才可以更快的知道它的用法，在需要的时候快速地熟悉它掌握它。