Rxjava基础篇
上一篇博客介绍了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
别看了,看也没有源代码。