上一篇博客介绍了Rxjava简单的使用和基本的流程，这次将会体验到Rxjava真正的威力。
线程控制-Scheduler（调度器）
Rxjava最大的特点就是异步，所以线程控制肯定是必不可少的。在Rxjava中线程控制是非常简单的。
主要的两个方法：
subscribeOn 指定OnSubscribe的call具体实现的线程。
observeOn指定Subscriber的回调发生在主线程
下面对其中的Rxjava中内置的几个比较常见的Scheduler进行简单的介绍。
- Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。
- Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。
- Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
- Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
- 另外， Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。
下面是一个简单的实例：模拟加载一张图片显示到ImageView上。
mImage = (ImageView) findViewById(R.id.image);
Observable.create(new Observable.OnSubscribe<Drawable>() {
@Override
public void call(Subscriber<? super Drawable> subscriber) {
Drawable drawable = getResources().getDrawable(drawableRes);
try {
Thread.sleep(7000);
//模拟图片加载的耗时操作，如果主线程休眠7s，会触发ANR。以此证明此处不是在主线程中执行的。
} catch (InterruptedException e) {
e.printStackTrace();
}
subscriber.onNext(drawable);
subscriber.onCompleted();
}
})
.subscribeOn(Schedulers.io())//指定subscribe发生在IO线程
.observeOn(AndroidSchedulers.mainThread())//指定Subscriber的回调发生在主线程中
.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) {
mImage.setImageDrawable(drawable);
}
});
运行之后，发现等待一段时间后，图片就可以正常显示在ImageView上。到这里，线程的调度的使用基本上就介绍完毕。下面开始介绍Rxjava中的交换。
交换
这里我们先直接通过例子来说明交换的作用和用法，然后在来介绍交换的API和具体使用方法。
考虑这样一种需求，从数据库中读取所有的用户数据，用R下java实现如下：eg1
bservable.create(new Observable.OnSubscribe<List<User>>() {
@Override
public void call(Subscriber<? super List<User>> subscriber) {
List<User> userList = null;
···
//从数据库获取用户表数据并赋给userList
···
subscriber.onNext(userList);
}
}).subscribe(new Action1<List<User>>() {
@Override
public void call(List<User> users) {
//获取到用户信息列表
}
});
然后，我们实际上只需要一个叫"xiaochu"的用户，然后我们代码就变成了下面的样式：eg2
Observable.create(new Observable.OnSubscribe<List<User>>() {
@Override
public void call(Subscriber<? super List<User>> subscriber) {
List<User> userList = null;
···
//从数据库获取用户表数据并赋给userList
···
subscriber.onNext(userList);
}
}).flatMap(new Func1<List<User>, Observable<User>>() {
@Override
public Observable<User> call(List<User> users) {
return Observable.from(users);
}
}).filter(new Func1<User, Boolean>() {
@Override
public Boolean call(User user) {
return user.getName().equals("xiaochu");
}
}).subscribe(new Action1<User>() {
@Override
public void call(User user) {
//xiaochu的数据
}
});
如果这个时候，我们需要使用xiaochu爸爸的数据，那么我们可以这么实现：eg3
Observable.create(new Observable.OnSubscribe<List<User>>() {
@Override
public void call(Subscriber<? super List<User>> subscriber) {
List<User> userList = null;
···
//从数据库获取用户表数据并赋给userList
···
subscriber.onNext(userList);
}
}).flatMap(new Func1<List<User>, Observable<User>>() {
@Override
public Observable<User> call(List<User> users) {
return Observable.from(users);
}
}).filter(new Func1<User, Boolean>() {
@Override
public Boolean call(User user) {
return user.getName().equals("xiaochu");
}
}).map(new Func1<User, User>() {
@Override
public User call(User user) {
//根据xiaochu的数据user从数据库查找出xiaochu的父亲user2
return user2;
}
}).subscribe(new Action1<User>() {
@Override
public void call(User user2) {
//拿到xiaochu爸爸的数据
}
});
从上面的示例中，可以看出来Rxjava强大的可扩展性吧。下面我们对其中的变化来进行具体的分析。
** 1. map() ** 事件对象的直接变换。
map()的使用在eg3就有具体的体现。
在Map参数中出现一个叫做Func1的类，与Action1非常相似，只不过他有返回值。与Action相似的是，Func也有多个（0-9）表示不同的参数的个数。
可以看到，通过map，我们将xiaochu的数据转换呈了xiaochu爸爸的数据。
**2. flatMap() ** 事件对象转换成Observable对象。
flatMap的使用在eg2就有具体的体现。通过flatMap，返回一个Observable对象，并且这个Observable对象并不是直接发送到了Subscriber的回调方法中。只是将传入的事件对象创建一个Observable,并将其激活，然后Observable汇入到同一个observable对象中，这个observable负责将这些事件统一交给了Subscriber的回调方法。这样做的好处就是将事件分成了两级，更加高效。
在eg2中，通过flatMap获取到了所有的users数据，并将他存储到Observable中。
**3. filter() ** 对Observable流程的数据进行过滤处理。返回值为false不会发送到Subscriber参照eg2的写法。eg2就过滤只剩下user名字为xiaochu的用户数据。
说明：filter的实现是lift()的变换处理，详细请参考lift变换部分。另外，filter()返回了一个新的Observable，因此若不是采用上面这种直接流方式，而是分步调用方式，需要将新返回的Observable赋给原来的Observable。
至于更加复杂的变化以及原理，这里就不做记录了。有兴趣的可以参考博客。http://gank.io/post/560e15be2dca930e00da1083
别看了，看也没有源代码。
