Rxjava入门二

通过上一节的学习我们知道了怎么样rxjava是什么(需要看一的朋友点这里Rxjava入门
一)，以及怎样利用rxjava打印输出字符串和数组。总结起来就是：Observable发射事件，subscribe订阅observable事件并迭代执行。

  但是我们知道我们平时所接触的事件流程远远没有这么简单。有时候我们需要把数据经过复杂的变换然后才能为我们所用。同样，rxjava的开发人员也已经为我们想到了这些。并且创造了一系列操作符为我们所用。

创建操作

  上一节我们已经接触了from和just两个创建方式，其实也是两个操作符。

除此之外还有Create、Range、Repeat、Start、Timer等方式

Create

  官方解释：使用一个函数从头开始创建一个Observable

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

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

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

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

示例代码:

1

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."); } });

  从例子我们可以看到。重新创建一个observable我们可以掌控订阅者对事件的执行方式。

Repeat

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

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

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

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





  关于创建的操作符就讲到这里，相关的其它方式可以在这里查看

变换操作符

  变换操作符的目的是讲数据变为我们需要的类型，常见的变换操作符有 Buffer、FlatMap、GroupBy、Map、Scan。

Map操作符

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

  使用map操作符，我们可以将数据转换后再进行发射





  如图所示，我们将每个数据*10然后输出数据会依次按照原始顺序*10然后输出.

下面我们举个例子

12 3 4 5 6 7 8 9 10 1112 13 14 15 16 17 18 19 20 2122 23 24

Observable.just("1010101010").map(new Func1<String, Integer>() { @Override public Integer call(String s) { return Integer.parseInt(s); } }).subscribe(new Subscriber<Integer>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Integer integer) { System.out.println(integer); } }); 06-03 14:55:13.927 358-358/? I/System.out: 1010101010

  我们可以看到，我们通过map操作符将String类型转换为int类型然后打印输出。map泛型第一个参数是原始数据类型，第二个是你需要的类型。在此不再过多赘述。

FlatMap操作符

  官方解释：FlatMap 将一个发射数据的Observable变换为多个Observables,然后将它们发射的数据合并 后放进一个单独的Observable

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

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





  官网文档意思是，利用flatmap可以将一个Observable发射的数据转换成多个Observable发射，然后将其发射的数据整合到一起，组成所需要的发射序列

我们来举个例子

Observable.just("a.b.c.d.e.f.g").flatMap(new Func1<string, observable>() {
@Override
public Observable call(String s) {
return Observable.from(s.split("\\."));
}
}).subscribe(new Subscriber() {
@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {

}

@Override
public void onNext(String s) {
System.out.println(s);
}
});

06-03 15:09:29.456 13428-13428/? I/System.out: a
06-03 15:09:29.456 13428-13428/? I/System.out: b
06-03 15:09:29.456 13428-13428/? I/System.out: c
06-03 15:09:29.456 13428-13428/? I/System.out: d
06-03 15:09:29.456 13428-13428/? I/System.out: e
06-03 15:09:29.456 13428-13428/? I/System.out: f
06-03 15:09:29.456 13428-13428/? I/System.out: g

  从代码中我们可以看到，我将字符串“a.b.c.d.e.f.g”用.符号分割开成长度为7的字符数组，然后利用Observable.from方式重新创建了数量为7的发射序列，然后依次进行发射。然后在依次subscribe中执行。

  变换操作符就先讲到这里，剩下的可在这里学习.

过滤操作符

  过滤操作符可以用于过滤和选择Observable发射的数据序列。常见的过滤操作符有 filter、takeLast、skip、skipLast、take、first、takeFirst、timeout等。过滤操作符看字面意思我想你已经知道差不多了。在这里我们不做重点解释。先举几个例子。

Filter

  官方解释：只发射通过了谓词测试的数据项。

Filter 操作符使用你指定的一个谓词函数测试数据项,只有通过测试的数据才会被发射。

示例代码：

12

Observable.just(1, 2, 3, 4, 5) .filter(new Func1<Integer, Boolean>() { @Override public Boolean call(Integer item) { return( item < 4 ); } }).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."); } });

输出

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

  剩余的操作符可以在这里这里继续学习。

  rxjava操作符类型还有很多，这里就不再列举了，有需要的可以看官方文档。这里还有一本中文版文档。需要的请点击阅读或下载。

Rxjava结合RxAndroid使用

调度器（Scheduler）

说到这里我们先提一下调度器的概念。

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

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

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

  Android是UI单线程的，所以在进行大量数据处理或者请求网络的时候我们经常要用到开启子线程异步处理数据。这时候我们就需要用到调度器了.

Rxjava使用调度器为我们提供了多种开启异步的方式。

如图：
rxjava_schedule.png

  看图得知，Schedulers.io( )操作符处理异步数据时是最适合我们使用的。

  指定一个观察者在哪个调度器上观察这个Observable，我们可以使用ObserveOn操作符

  指定Observable自身在哪个调度器上执行我们可以使用SubscribeOn操作符

RxAndroid

  RxAndroid是Rxjava的扩展，是由JakeWharton编写，目的是使Android可以优雅地处理异步请求并且完美融合Rxjava使用。RxAndroid为我们提供了一个AndroidSchedulers.mainThread()调度器，我们可以在observer中自由切换线程。

我们先写个例子：

Observable.just("one", "two", "three", "four", "five")
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(/* an Observer */);

这样Observable会在子线程里处理事件，处理结果会发送到onnext回调方法中去。

  今天就先说这么多了，这一篇主要讲了Rxjava的操作符和RxAndroid使用，下一章将结合Retrofit使用。