Android RxJava详解
这两天闲来无事就想着学习一下当前比较流行的框架,看到网上很多大神推荐RxJava
和Retrofit的混合使用,就想着自己也学习一下封装一下,于是就查阅了很多文章总是感
觉缺少点什么,直到碰到了他的文章(链接在文章末尾)才有一种恍然大悟的感觉,在此做
一个总结。
响应式代码的基本组成部分是Observables和Subscribers(事实上Observer才是最小的构建块,但实践中使用最多的是Subscriber,因为Subscriber才是和Observables的对应的。)。Observable发送消息,而Subscriber则用于消费消息。
-
RxJava 基本概念
- Observable (可观察者,即被观察者)
- Observer (观察者)
- subscribe (订阅)、事件
- Scheduler 调度器,相当于线程控制器
Observable 和Observer 通过 subscribe() 方法实现订阅关系,从而 Observable 可以在需要的时候发出事件来通知 Observer。
-
RxJava的实现流程
-
创建Observer(观察者)
Observer 即观察者,它决定事件触发的时候将有怎样的行为.RxJava中的Observer的实现方式:
Observer<String> observer = new Observer<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!"); } };
Subscriber是对Observe的抽象类,RxJava对其实现了一些扩展,但是使用方式基本一致:
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!"); } };
Observer也总是会先被转换成一个Subscriber再使用,不同的是Subscriber中又增加了两个方法onStart() 和 UnSubscriber();
-
创建Observable(被观察者)
Observable即被观察者,它决定什么时候触发事件以及触发怎样的事件。RxJava 使用create()方法来创建一个Observable,并为它定义事件触发规则:
Observable observable = Observable.create(new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { subscriber.onNext("Hello"); subscriber.onNext("Hi"); subscriber.onNext("Aloha"); subscriber.onCompleted(); } });
这里传入了一个OnSubscribe对象作为参数。OnSubscribe会被存储在返回的 Observable对象中,它的作用相当于一个计划表,当Observable被订阅的时候,OnSubscribe的call()方法会自动被调用,事件序列就会依照设定依次触发(对于上面的代码,就是观察者Subscriber 将会被调用三次 onNext() 和一次 onCompleted())。这样,由被观察者调用了观察者的回调方法,就实现了由被观察者向观察者的事件传递.
-
创建Subscribe(订阅)
创建了 Observable 和 Observer 之后,再用 subscribe() 方法将它们联结起来:
observable.subscribe(observer); // 或者: observable.subscribe(subscriber);
-
以上的代码也可以这么写:
Observable.create(new OnSubscribe<Drawable>() {
@Override
public void call(Subscriber<? super Drawable> subscriber) {
Drawable drawable = getTheme().getDrawable(drawableRes));
subscriber.onNext(drawable);
subscriber.onCompleted();
}
}).subscribe(new Observer<Drawable>() {
@Override
public void onNext(Drawable drawable) {
imageView.setImageDrawable(drawable);
}
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
Toast.makeText(activity, "Error!", Toast.LENGTH_SHORT).show();
}
});
-
Scheduler--线程控制
在不指定线程的情况下,RxJava遵循的是线程不变的原则,即:在哪个线程调用 subscribe(),就在哪个线程生产事件;在哪个线程生产事件,就在哪个线程消费事件。如果需要切换线程,就需要用到Scheduler(调度器),下面是Scheduler的API:
- 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 主线程运行。
有了这几个 Scheduler ,就可以使用 subscribeOn() 和 observeOn() 两个方法来对线程进行控制了。
subscribeOn(): 指定subscribe()所发生的线程,即 Observable.OnSubscribe被激活时所处的线程。或者叫做事件产生的线程。
observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。
Demo:
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);
}
});
Operators:RxJava提供了对事务序列进行变换的功能,这是RxJava的核心功能之一。变换就是将事件序列中的对象或整个序列进行加工处理,转换成不同的事件或事件序列。
-
from操作符可以转换Future、Iterable和数组。对于Iterable和数组,产生的Observable会发射Iterable或数组的每一项数据。
Integer[] items = { 0, 1, 2, 3, 4, 5 }; Observable myObservable = Observable.from(items); myObservable.subscribe( new Action1<Integer>() { @Override public void call(Integer item) { System.out.println(item); } }, new Action1<Throwable>() { @Override public void call(Throwable error) { System.out.println("Error encountered: " + error.getMessage()); } }, new Action0() { @Override public void call() { System.out.println("Sequence complete"); } } );
-
Just:Just类似于From,但是From会将数组或Iterable的素具取出然后逐个发射,而Just只是简单的原样发射,将数组或Iterable当做单个数据。
注意:如果你传递null给Just,它会返回一个发射null值的Observable。不要误认为 它会返回一个空Observable(完全不发射任何数据的Observable),如果需要空 Observable你应该使用Empty操作符。
Observable.just(1, 2, 3) .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."); } });
-
map():Map操作符对原始Observable发射的每一项数据应用一个你选择的函数,然后返回一个发射这些结果的Observable
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); } });
map() 方法将参数中的 String 对象转换成一个 Bitmap 对象后返回,而在经过 map() 方法后,事件的参数类型也由 String转为了 Bitmap。这种直接变换对象并返回的,是最常见的也最容易理解的变换。不过 RxJava 的变换远不止这样,它不仅可以针对事件对象,还可以针对整个事件队列,这使得 RxJava 变得非常灵活。
-
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);
flatMap() 和 map() 有一个相同点:它也是把传入的参数转化之后返回另一个对象。但需要注意,和 map()不同的是, flatMap() 中返回的是个 Observable 对象,并且这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。 flatMap() 的原理是这样的:1. 使用传入的事件对象创建一个 Observable 对象;2. 并不发送这个 Observable, 而是将它激活,于是它开始发送事件;3. 每一个创建出来的 Observable 发送的事件,都被汇入同一个 Observable ,而这个 Observable 负责将这些事件统一交给 Subscriber 的回调方法。这三个步骤,把事件拆成了两级,通过一组新创建的 Observable 将初始的对象『铺平』之后通过统一路径分发了下去。而这个『铺平』就是 flatMap() 所谓的 flat。
-
throttleFirst():在每次事件触发后的一定时间间隔内丢弃新的事件。常用作去抖动过滤,例如按钮的点击监听器:
RxView.clickEvents(button) //RxBinding代码 .throttleFirst(500, TimeUnit.MILLISECONDS) //设置防抖间隔为500ms .subscribe(subscriber);
建议大家多去理解一下RxJava的变换,这是个难点和功能点。