Android开发——关于RxJava的知识总结
0. 前言
RxJava在主页上的介绍：
//a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
//一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库。大家都知道异步代码经常会既难写也难被读懂。这时候RxJava的优势就来了，随着程序逻辑变得越来越复杂，它依然能够保持简洁。初学RxJava只要把握两点，观察者模式和异步。
1. 观察者模式
被观察者：被监视的对象，当其某个状态改变的时候会通知观察者；
观察者：监视着被观察者的行为，当被观察者某个状态改变的时候会通知观察者，观察者会执行对应的操作；
订阅：将观察者和被观察者建立联系。
这里举一个Android中常见的例子，点击Button后触发OnClickListener中的onClick()事件。在这个例子中被观察者就是Button、观察者：OnClickListener、订阅的逻辑就是setOnClickListener的过程。
2. RxJava
2.1 RxJava的简单示例
RxJava也是基于观察者模式来组建自己的程序逻辑的，就是构建被观察者（Observable），观察者（Observer/Subscriber），然后通过Subscribe()建立两者的订阅关系。
直接看代码比较容易懂：
//功能为按顺序打印一些字符串
//别忘记添加依赖compile 'io.reactivex:rxjava:1.0.14' compile 'io.reactivex:rxandroid:1.0.1'
//创建一个观察者
Observer<String> observer = new Observer<String>() {
@Override
public void onCompleted() {
Log.i(TAG, "Completed");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "Error");
}
@Override
public void onNext(String s) {
Log.i(TAG, s);
}
};
//使用Observable.create()创建被观察者
Observable observable1 = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello");
subscriber.onNext("Wrold");
subscriber.onCompleted();
}
});
//订阅
observable1.subscribe(observer);2.2 一些拓展
2.2.1 关于Observable的简单写法
上面介绍了Observable的一种最正规的创建方式，当然也可以偷懒写成下面这三种的样子，效果是一样的：
Observable observable2 = Observable.just("Hello", "World");
String [] words = {"Hello", "World"};
Observable observable3 = Observable.from(words);
List<String> list = new ArrayList<String>();
list.add("Hellow");
list.add("Wrold");
Observable observable4 = Observable.from(list);2.2.2 关于Observer的扩展
除了Observer接口之外，RxJava 还内置了一个实现了Observer的抽象：Subscriber。
Subscriber对Observer接口进行了一些扩展，但他们的基本使用方式是完全一样的。而且不仅基本使用方式一样，实质上在 RxJava 的 subscribe 过程中，Observer也总是会先被转换成一个Subscriber再使用。不仅基本使用方式一样，实质上在 RxJava 的 subscribe 过程中，Observer也总是会先被转换成一个Subscriber再使用。
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onNext(String s) {
Log.d(tag, "Item: " + s);
}
@Override
public void onCompleted() {
Log.d(tag, "Completed!");
}
@Override
public void onError(Throwable e) {
Log.d(tag, "Error!");
}
};
它们的区别对于使用者来说主要有两点：
（1）onStart(): 这是Subscriber增加的方法。它会在 subscribe 刚开始，而事件还未发送之前被调用。
onStart()可以用于做一些准备工作，例如数据的清零或重置。这是一个可选方法，默认情况下它的实现为空。需要注意的是，onStart()总是在 subscribe 所发生的线程被调用，而不能指定线程。因此主线程中UI的准备逻辑，就不适宜在这里写了。
（2）unsubscribe(): 这是Subscriber所实现的另一个接口Subscription的方法，用于取消订阅。
在这个方法被调用后，Subscriber将不再接收事件。一般在这个方法调用前，可以使用isUnsubscribed()先判断一下状态。unsubscribe()这个方法很重要，可以较少内存泄露的风险。因为在subscribe()之后，Observable会持有Subscriber的引用，这个引用如果不能及时被释放，将有内存泄露的风险。
2.2 Action
有时候我们不需要onError和onCompleted这两个方法的观察者回调，而只需要onNext()方法，这时就可以使用Action来代替Subscriber。如下所示：
Observable.just("Hello", "World").subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.i(TAG, s);
}
});2.2.1 什么是Action
Action的使用为我们减少了不必要的代码，使得写出的代码看上去更加得简洁。Action是RxJava 的一个接口，常用的有Action0和Action1。
（1）Action0： 它只有一个方法 call()，这个方法是无参无返回值的；由于 onCompleted() 方法也是无参无返回值的，因此 Action0 可以被当成一个包装对象，将 onCompleted() 的内容打包起来将自己作为一个参数传入 subscribe() 以实现不完整定义的回调。
（2）Ation1：它同样只有一个方法 call(T param)，这个方法也无返回值，但有一个参数；与 Action0 同理，由于 onNext(T obj) 和 onError(Throwable error) 也是单参数无返回值的，因此 Action1 可以将 onNext(obj)和 onError(error) 打包起来传入 subscribe() 以实现不完整定义的回调。
Observable observable = Observable.just("Hello", "World");
//处理onNext()中的内容
Action1<String> onNextAction = new Action1<String>() {
@Override
public void call(String s) {
Log.i(TAG, s);
}
};
//处理onError()中的内容
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
}
};
//处理onCompleted()中的内容
Action0 onCompletedAction = new Action0() {
@Override
public void call() {
Log.i(TAG, "Completed");
}
};
//使用 onNextAction 来定义 onNext()
Observable.just("Hello", "World").subscribe(onNextAction);
//使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
Observable.just("Hello", "World").subscribe(onNextAction, onErrorAction);
//使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
Observable.just("Hello", "World").subscribe(onNextAction, onErrorAction, onCompletedAction);//现在有个疑问，为什么使用Action也能达到使用Subscriber的结果？
//进subscribe(Action1 onNext)的源码看看。如下图所示
//原来就是把Action对象转化成对应的Subscriber对象了
//这样就不难理解为什么可以使用Action来代替Subscriber了。
2.3 map/flatMap
在使用map之前还要说一个接口：Func1。
Func1和上一篇提到的Action1相似。Func1和Action1的区别在于，Func1包装的是有返回值的方法。
接下来就是map的用法，看代码更直观一点。
//得到多个Student对象中的name，保存到nameList中
Observable.just(student1, student2, student2)
//使用map进行转换，参数1：转换前的类型，参数2：转换后的类型
.map(new Func1<Student, String>() {
@Override
public String call(Student i) {
String name = i.getName();
return name;
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
nameList.add(s);
}
});flatMap是一个比较难理解的一个转换，在这里先假设一个需求，需要打印多个Student所学的课程。这跟之前获取Student的name又不同了，一个Student类中只有一个name，但是一个Student却有多门课程Course。
* 学生类
*/
class Student {
private String name;//姓名
private List<Course> coursesList;//所修的课程
...
}
/**
* 课程类
*/
class Course {
private String name;//课程名
private String id;
...
}
List<Student> students = new ArrayList<Student>();
students.add...
...
Action1<List<Course>> action1 = new Action1<List<Course>>() {
@Override
public void call(List<Course> courses) {
//遍历courses，输出cuouses的name
for (int i = 0; i < courses.size(); i++){
Log.i(TAG, courses.get(i).getName());
}
}
};
Observable.from(students)
.map(new Func1<Student, List<Course>>() {
@Override
public List<Course> call(Student student) {
//返回coursesList
return student.getCoursesList();
}
})
.subscribe(action1);
List<Student> students = new ArrayList<Student>();
students.add...
...
Observable.from(students)
.flatMap(new Func1<Student, Observable<Course>>() {
@Override
public Observable<Course> call(Student student) {
return Observable.from(student.getCoursesList());
}
})
.subscribe(new Action1<Course>() {
@Override
public void call(Course course) {
Log.i(TAG, course.getName());
}
});
（1）使用传入的事件对象创建一个Observable对象；
（2）并不发送这个Observable，而是**它，使其开始发送事件；
（3）每一个创建出来的Observable发送的事件，都被汇入到同一个Observab，而这个Observable负责将这些事件统一交给Subscriber的回调方法。
经过了这三步，就把事件拆分成了两级，通过一组新创建的Observable将初始的对象铺平之后通过统一路径分发了下去。
关于RxJava的其他操作符，比如filter，take，skip等，有兴趣的可以在这个训练项目中进行熟悉练习。
2.4 线程控制
RxJava在不指定线程的情况下，发起时间和消费时间默认使用当前线程。因此若是map中有耗时的操作，都是在主线程中进行的，这样会导致主线程拥塞。
Scheduler：线程控制器，就是来解决上述问题的，它可以指定每一段代码在什么样的线程中执行。
模拟一个需求：新的线程发起事件，在主线程中消费，这就用到了subscribeOn()和observeOn()方法来指定发生的线程和消费的线程。前者指定事件产生的线程，只能指定一次，如果指定多次则以第一次为准。后者指定Subscriber所运行的线程，或者说是事件消费的线程，可以指定多次，每次指定完都会在下一步一直生效。下面是这个需求的实现代码，以及关于这两个方法参数，RxJava为我们提供的几个Scheduler。
Observable.just("Hello", "Word")
.subscribeOn(Schedulers.newThread())//指定 subscribe() 发生在新的线程
.observeOn(AndroidSchedulers.mainThread())// 指定 Subscriber 的回调发生在主线程
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.i(TAG, s);
}
});
//1.Schedulers.immediate()：直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。
//2.Schedulers.newThread()：总是启用新线程，并在新线程执行操作。
//3.Schedulers.io()： I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。
//和 newThread() 区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程
//因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，避免创建不必要的线程。
//4.Schedulers.computation()：计算所使用的 Scheduler。
//这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。
//使用CPU核数大小的固定线程池，I/O 操作如果被放在这里，会因为I/O等待的时间而浪费 CPU。
//5.AndroidSchedulers.mainThread()：它指定的操作将在 Android 主线程运行。2.5 多次切换线程
上面只是对事件的发起和消费制定了线程。如果中间有map之类的操作呢？是否可以实现发起的线程在新线程中，map的处理在IO线程，最后的消费在主线程中呢？
可以看到observeOn()被调用了两次，分别指定了map的处理的现场和消费事件show(s)的线程。
Observable.just("Hello", "Wrold")
.subscribeOn(Schedulers.newThread())//指定在新的线程中发起
.observeOn(Schedulers.io()) //指定在io线程中处理
.map(new Func1<String, String>() {
@Override
public String call(String s) {
return handleString(s); //处理数据
}
})
.observeOn(AndroidSchedulers.mainThread())//指定在主线程中处理
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
show(s); //消费事件
}
});参考文献：
http://gank.io/post/560e15be2dca930e00da1083
http://www.jianshu.com/p/19cac3c5b106
上一篇： 女人腰疼吃什么好，这些健康食物不能错过
下一篇： 随机数生成