前言
最近看到各个公司的项目都在用RxJava，意识到必须要学习一下这个库了，所以找到了“扔物线”所发表的文章<a href="http://gank.io/post/560e15be2dca930e00da1083#toc_6">“给 Android 开发者的 RxJava 详解”</a>,以下是个人在学习过程中的一些笔记与摘要，在此分享。
RxJava有四个概念
1、Observable---被观察者
2、Observer---观察者
3、subscribe---订阅
4、事件
RxJava中有三个主要回调
1、onNext() ----事件，多个onNext()会组成事件队列执行。
2、onCompleted() ----事件队列完结，当没有新的onNext事件发出时，就会调用onCompleted()。
3、onError() ----事件队列异常，当事件队列执行过程发生异常就会调用该方法，并且不会再执行后续的onNext()事件。
Observer和Subscriber
1、Subscriber是Observer的子类，使用方式基本完全一样，并且Observer在使用的时候，也会被先转换成为一个Subscriber后执行。
2、Subscriber和Observer的区别主要有两点：
1）onStart()：Subscriber所新增的方法，在subscribe订阅刚开始，事件发送之前被调用。但是该方法无法指定线程，它会在subscribe发生的线程被执行。
2）unsubscribe(): 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。在这个方法被调用后，Subscriber 将不再接收事件。一般在这个方法调用前，可以使用 isUnsubscribed() 先判断一下状态。 unsubscribe() 这个方法很重要，因为在 subscribe() 之后， Observable 会持有 Subscriber 的引用，这个引用如果不能及时被释放，将有内存泄露的风险。所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop() 等方法中）调用 unsubscribe() 来解除引用关系，以避免内存泄露的发生。
基本代码示例
private void loadImg() {
final int resId = R.mipmap.ic_launcher;
Observable.create(new Observable.OnSubscribe<Drawable>() {
@TargetApi(Build.VERSION_CODES.LOLLIPOP)
@Override
public void call(Subscriber<? super Drawable> subscriber) {
Drawable drawable = getTheme().getDrawable(resId);
subscriber.onNext(drawable);
subscriber.onCompleted();
}
}).subscribe(new Observer<Drawable>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
Toast.makeText(MainActivity.this, "error", Toast.LENGTH_SHORT).show();
}
@Override
public void onNext(Drawable drawable) {
mImageView.setImageDrawable(drawable);
}
});
}
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
Log.d("MainActivity", "事件队列结束");
}
@Override
public void onError(Throwable e) {
Log.d("MainActivity", "事件队列出现异常");
}
@Override
public void onNext(String s) {
Log.d("MainActivity", s);
}
};
Observable observable = Observable.from(new String[]{"休眠开始", "休眠结束"}).just("新的吧");
observable.subscribe(subscriber);
线程控制--Scheduler
在不指定线程的情况下， RxJava 遵循的是线程不变的原则，即：在哪个线程调用 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到 Scheduler （调度器）。
一、RxJava内置的几个Scheduler
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 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
5、另外， Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。
二、使用方式。
有了这几个 Scheduler ，就可以使用 subscribeOn() 和 observeOn() 两个方法来对线程进行控制了。 * subscribeOn(): 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。 * observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。
示例代码如下：
Observable.just(1, 2, 3, 4)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer number) {
Log.d(tag, "number:" + number);
}
});
变换
1、map()：一对一的变换
示例代码：
Observable.just("images/logo.png") // 输入类型 String
.map(new Func1<String, Bitmap>() {
@Override
public Bitmap call(String filePath) { // 参数类型 String
return getBitmapFromPath(filePath); // 返回类型 Bitmap
}
})
.subscribe(new Action1<Bitmap>() {
@Override
public void call(Bitmap bitmap) { // 参数类型 Bitmap
showBitmap(bitmap);
}
});
上面示例代码可以看到，变换的作用就是在结果被回调之前，对输入的数据进行处理，再输出到observer。
上述代码我们看到一个Func1，Func和Action其实很相似，只是Func是带有返回值的，同样的Func也有多个FuncX。
map()所实现的变换只针对一对一的情况，即一个输入得到一个结果。
2、flatMap():一对多变换
示例代码：
Student[] students = ...;
Subscriber<Course> subscriber = new Subscriber<Course>() {
@Override
public void onNext(Course course) {
Log.d(tag, course.getName());
}
...
};
Observable.from(students)
.flatMap(new Func1<Student, Observable<Course>>() {
@Override
public Observable<Course> call(Student student) {
return Observable.from(student.getCourses());
}
})
.subscribe(subscriber);
可以看到，上面的代码将每一个student所包含该的多个课程全部取了出来，这就是一对多的情况，这种情况就可以通过flatMap来实现。
flatMap所返回的是一个Observable对象，但这个Observable并不会直接发送到Subscriber的回调中，而是将这个observable所发送出来的事件汇集，构成一个新的observable，再由这个新的observable统一的将事件交给Subscriber的回调。
doOnSubscribe()
前面提到Subscriber中有一个onStart()方法，这个方法可以做一些初始化的操作，是在订阅开始后，事件发送前被调用的。但是它并不能指定线程，它永远只会在subscribe的线程中被执行。那当我们的初始化操作需要指定线程时就需要使用到doOnSubscribe()来做了。
示例代码：
Observable.create(onSubscribe)
.subscribeOn(Schedulers.io())
.doOnSubscribe(new Action0() {
@Override
public void call() {
progressBar.setVisibility(View.VISIBLE); // 需要在主线程执行
}
})
.subscribeOn(AndroidSchedulers.mainThread()) // 指定主线程
.observeOn(AndroidSchedulers.mainThread())
.subscribe(subscriber);
默认情况下，doOnSubscribe()也将在subscribe所发生的线程中被执行，但是如果doOnSubscribe()之后有一个subscribeOn()的话，那么它将会在它后面最近的那个subscribeOn()所指定的线程中执行。
