RxJava1 入门笔记

RxJava什么功能？

RxJava封装了异步操作，如同AsyncTask和Handler一样，但是功能更强大。

一、最简单的用法，RxJava是基于观察者模式的，而观察者模式，按我的理解就是，Observable 是事件源，

订阅了 Observable的是观察者（也叫订阅者），事件源 会持有 所有订阅者的引用，所以在产生事件时，事件源可以地调用

所有 订阅者的特定方法，这看起来就是 订阅者及时对事件源产生的事件作出了响应。

//先创建一个Observable<T>即 事件源（被观察者），在call()方法中调用它的观察者的onNext("Hello")、onCompleted()等方法
//当 Observable 被订阅的时候，OnSubscribe 的 call() 方法会自动被调用
final Observable<String> eventSource = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("first");
subscriber.onNext("second");
subscriber.onNext("third");
subscriber.onCompleted();
}
});

//创建一个观察者，当事件源 Observable 派发事件时，onCompleted()、onNext(T t)等方法会被 Observable 调用
final Observer<String> observer = new Observer<String>() {
@Override
public void onCompleted() {
Log.i(TAG, "onCompleted: ");
}

@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: ");
}

@Override
public void onNext(String s) {
Log.i(TAG, "onNext: " + s);
}
};

//点击按钮，observer 订阅事件源 Observable
btn.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
eventSource.subscribe(observer);
}
});

RxJava 提供了一些简化代码的方法，如 Observable.just()、Observable.from() 等方法，以下三种写法完全等价：

Observable<String> eventSource = Observable.just("first", "second", "third");

Observable<String> eventSource = Observable.from(new String[]{"first", "second", "third"});

Observable<String> eventSource = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("first");
subscriber.onNext("second");
subscriber.onNext("third");
subscriber.onCompleted();
}
});

二、 RxJava的核心功能之一是可以轻易的切换线程，使用Schedules类的几个静态方法：

Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。

Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。

Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 i
4000
o() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。

Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 - computation() 中，否则 I/O 操作的等待时间会浪费 CPU。

引进依赖库 “RxAndroid”, 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。

final Observable<String> eventSource = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("first");
subscriber.onNext("second");
subscriber.onNext("third");
subscriber.onCompleted();
}

}).subscribeOn(Schedulers.io())    //订阅的方法（加载任务）在io线程中进行
.observeOn(AndroidSchedulers.mainThread());//观测者回调的方法（更新UI）在主线程中执行

三、 RxJava核心功能之二就是变换数据源，数据源由一个 Observable （被观察的对象）发出，数据的类型由Observable 的泛型指定。假设现在有一个存放着一组图片 url 的 String 数组，但是我希望 Observable 发出的数据直接是 url 对应的 Bitmap 对象，这样观察者在收到数据时就可以直接得到 Bitmap 对象，那么可以用 RxJava 提供的 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 Observer<Bitmap>() {
@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {

}

@Override
public void onNext(Bitmap bitmap) {
showBitmap(bitmap);
}
});

有时候我们并不关心观察者获取到数据之后出现error的情况，也不希望在 onCompleted() 中做什么事，就像上面的例子，可以用一个Action1 代替一个完整的 Observer 对象：

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);
}
});

RxJava 的用途很多，这里先涉及了几个基本用法，下一篇将利用 RxJava 实现 EventBus 的功能，也就是常说的RxBus。

参考：网上一篇很清晰详细的 RxJava 入门教程