注: 本文只适用于自我复习,初学者请移步 链接
基础概念复习
1) 创建 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!"); } };
除了 Observer
接口之外,RxJava 还内置了一个实现了 Observer
的抽象类:Subscriber
。 Subscriber
对 Observer
接口进行了一些扩展,但他们的基本使用方式是完全一样的
不仅基本使用方式一样,实质上,在 RxJava 的 subscribe 过程中,Observer
也总是会先被转换成一个 Subscriber
再使用。所以如果你只想使用基本功能,选择 Observer
和 Subscriber
是完全一样的。它们的区别对于使用者来说主要有两点:
1. onStart()
: 这是 Subscriber
增加的方法。它会在 subscribe 刚开始,而事件还未发送之前被调用,可以用于做一些准备工作,例如数据的清零或重置。这是一个可选方法,默认情况下它的实现为空。需要注意的是,如果对准备工作的线程有要求(例如弹出一个显示进度的对话框,这必须在主线程执行), onStart()
就不适用了,因为它总是在 subscribe 所发生的线程被调用,而不能指定线程。要在指定的线程来做准备工作,可以使用 doOnSubscribe()
方法,具体可以在后面的文中看到。
2. unsubscribe()
: 这是 Subscriber
所实现的另一个接口 Subscription
的方法,用于取消订阅。在这个方法被调用后,Subscriber
将不再接收事件。一般在这个方法调用前,可以使用 isUnsubscribed()
先判断一下状态。 unsubscribe()
这个方法很重要,因为在 subscribe()
之后, Observable
会持有 Subscriber
的引用,这个引用如果不能及时被释放,将有内存泄露的风险。所以最好保持一个原则:要在不再使用的时候尽快在合适的地方(例如 onPause()
onStop()
等方法中)调用 unsubscribe()
来解除引用关系,以避免内存泄露的发生
2) 创建 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()
)。这样,由被观察者调用了观察者的回调方法,就实现了由被观察者向观察者的事件传递,即观察者模式。
create()
方法是 RxJava 最基本的创造事件序列的方法。基于这个方法, RxJava 还提供了一些方法用来快捷创建事件队列,例如:
* just(T...)
: 将传入的参数依次发送
Observable observable = Observable.just("Hello", "Hi", "Aloha"); // 将会依次调用: // onNext("Hello"); // onNext("Hi"); // onNext("Aloha"); // onCompleted();
* from(T[])
/ from(Iterable<? extends T>)
: 将传入的数组或 Iterable
拆分成具体对象后,依次发送出来。
String[] words = {"Hello", "Hi", "Aloha"}; Observable observable = Observable.from(words); // 将会依次调用: // onNext("Hello"); // onNext("Hi"); // onNext("Aloha"); // onCompleted();
1) Scheduler 的 API (一)
在RxJava 中,Scheduler
——调度器,相当于线程控制器,RxJava 通过它来指定每一段代码应该运行在什么样的线程。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 主线程运行。
有了这几个 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); } });
上面这段代码中,由于 subscribeOn(Schedulers.io())
的指定,被创建的事件的内容 1
、2
、3
、4
将会在 IO 线程发出;而由于 observeOn(AndroidScheculers.mainThread()
) 的指定,因此 subscriber
数字的打印将发生在主线程 。事实上,这种在 subscribe()
之前写上两句 subscribeOn(Scheduler.io())
和 observeOn(AndroidSchedulers.mainThread())
的使用方式非常常见,它适用于多数的 『后台线程取数据,主线程显示』的程序策略。
而前面提到的由图片 id 取得图片并显示的例子,如果也加上这两句:
int drawableRes = ...; ImageView imageView = ...; Observable.create(new OnSubscribe<Drawable>() { @Override public void call(Subscriber<? super Drawable> subscriber) { Drawable drawable = getTheme().getDrawable(drawableRes)); subscriber.onNext(drawable); subscriber.onCompleted(); } }) .subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程 .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程 .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(); } });
首先看一个 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); } });
这里出现了一个叫做 Func1
的类。它和 Action1
非常相似,也是 RxJava 的一个接口,用于包装含有一个参数的方法。 Func1
和 Action
的区别在于, Func1
包装的是有返回值的方法。另外,和 ActionX
一样, FuncX
也有多个,用于不同参数个数的方法。FuncX
和 ActionX
的区别在 FuncX
包装的是有返回值的方法。
可以看到,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。
常用模板