RXjava学习笔记

前言

本文主要是自己学习RxJava的笔记整理

学习的链接

http://gank.io/post/560e15be2dca930e00da1083#toc_1

Rx介绍

ReactiveX是Reactive extensions的缩写，一般简写为Rx。最初是LINQ的一个扩展，由微软的架构师erik meijer领导的团队开发。微软给的定义是，Rx是一个函数库，让开发者可以利用可观察序列和LINQ风格查询操作符来编写异步和基本事件的程序，使用Rx，开发者可以用observables表示异步数据流，用LINQ操作符查询异步数据流，用scheduler参数化异步数据流的并发处理，Rx可以这样定义：Rx=Observable+LINQ+scheduler。

RxJava是什么

RxJava是一个响应式编程框架，采用的是观察者模式。也可以用一个词来概况，那就是：异步。

有很多术语可用于描述这种异步编程和设计模式：一个观察者订阅一个可观察对象。通过调用观察者的方法，observable发射数据活通知给它的观察者。

RxJava在github上的地址是https://github.com/ReactiveX/RxJava

在主页中，RxJava 的自我介绍是”a library for composing asychronous and event-based programs using observable sequences for the Java VM”（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）。

API介绍和原理解析

概念：扩展的观察者模式

RxJava的异步实现，是通过一种扩展的观察者模式来实现的。但是有一点明显的不同，就是如果一个observable没有任何的subscriber，那么这个observable是不会发出任何事件的。

观察者模式：

有时候也被称为发布/订阅模式，观察者模式定义了一种一对多的依赖关系，让多个观察者对象同时监听某一个主题对象。这个主题对象在状态发生改变时，会通知所有观察者对象，是他们能够自动更新自己。

观察者模式面向的需求是：A 对象（观察者）对 B 对象（被观察者）的某种变化高度敏感，需要在 B 变化的一瞬间做出反应。举个例子，新闻里喜闻乐见的警察抓小偷，警察需要在小偷伸手作案的时候实施抓捕。在这个例子里，警察是观察者，小偷是被观察者，警察需要时刻盯着小偷的一举一动，才能保证不会漏过任何瞬间。程序的观察者模式和这种真正的『观察』略有不同，观察者不需要时刻盯着被观察者（例如 A 不需要每过 2ms 就检查一次 B 的状态），而是采用注册(Register)或者称为订阅(Subscribe)的方式，告诉被观察者：我需要你的某某状态，你要在它变化的时候通知我。 Android 开发中一个比较典型的例子是点击监听器 OnClickListener 。对设置 OnClickListener 来说， View 是被观察者，

OnClickListener 是观察者，二者通过 setOnClickListener() 方法达成订阅关系。订阅之后用户点击按钮的瞬间，Android Framework 就会将点击事件发送给已经注册的 OnClickListener 。采取这样被动的观察方式，既省去了反复检索状态的资源消耗，也能够得到最高的反馈速度。当然，这也得益于我们可以随意定制自己程序中的观察者和被观察者，而警察叔叔明显无法要求小偷『你在作案的时候务必通知我』。

OnClickListener 的模式：通过 setOnClickListener() 方法，Button 持有 OnClickListener 的引用（这一过程没有在图上画出）；当用户点击时，Button 自动调用 OnClickListener 的 onClick() 方法。另外，如果把这张图中的概念抽象出来（Button -> 被观察者、OnClickListener -> 观察者、setOnClickListener() -> 订阅，onClick() -> 事件），就由专用的观察者模式（例如只用于监听控件点击）转变成了通用的观察者模式。

RxJava的观察者模式

RxJava有是个基本概念：Observable (可观察者，即被观察者)、 Observer (观察者)、 subscribe (订阅)、事件。Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知 Observer。

与传统观察者模式不同， RxJava 的事件回调方法除了普通事件 onNext() （相当于 onClick() / onEvent()）之外，还定义了两个特殊的事件：onCompleted() 和 onError()。

onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。

onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。

在一个正确运行的事件序列中, onCompleted() 和 onError() 有且只有一个，并且是事件序列中的最后一个。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。

基本实现

创建观察者Observer

observer，它决定了事件触发的时候将有什么行为。RxJava 中的observer接口的实现方式：

Observer<String> observer = new Observer<String>() {
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted: ");
}

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

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

RxJava 中还有一个subscriber类是对observer类进行了扩展

Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onStart() {
super.onStart();
}
@Override
public void onCompleted() {
Log.d(TAG, "onCompleted: ");
}

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

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

这两个类不单单使用方式是一样的，实质上，在RxJava的subscriber过程中，observer也总是会先被转换成一个subscriber再使用。所以如果你只想使用基本功能，选择observer和subscriber是完全一样的。他们区别对于使用者来说，主要是两点：

onStart()：这是subscriber增加的方法。他会在subscriber刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如：数据的清零或者重置、这是一个可选方法，默认情况下它的实现为空。需要注意的是，如果对准备工作的线程有要求（例如弹出一个显示进度的对话框，这必须在主线程执行）， onStart() 就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法，具体可以在后面的文中看到。

unsubscriber()：这是subscriber所实现的另一个接口subscription的方法，用于取消订阅。在这个方法被调用后，subscriber将不再接受事件。一般在这个方法调用之前，可以使用.isUnsubscribed()来判断一下当前状态。 unsubscribe() 这个方法很重要，因为在 subscribe() 之后， Observable 会持有 Subscriber 的引用，这个引用如果不能及时被释放，将有内存泄露的风险。所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop() 等方法中）调用 unsubscribe() 来解除引用关系，以避免内存泄露的发生。

Observable 被观察者的创建

Observable ，它决定什么时候触发事件以及触发怎样的事件。实现方法直接调用observable.create：

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("one");
subscriber.onNext("two");
subscriber.onNext("three");
subscriber.onCompleted();
}
});

observable绑定完成observer后，即：observable.subscribe(observer)，有事件发生，调用call方法后，会生成subscriber对象，subscriber对象调用onnext，oncomplete方法，也就是相当于直接调用了绑定的observer的onnext，oncomplete方法。

create方法是最基本的创建事件序列的方法。基于这个方法，RxJava还提供了一些用来快捷创建事件队列：

just(T….)：将传入的参数依次发送出来。

just将单个数据转换为发射那个数据的observable，注意：如果你传递null给just，它会返回一个发射null值的observable。不要误认为它会返回一个空的observable（完全不发射任何数据的observable），如果需要空observable，你应该使用empty操作符。

/**
* 将依次调用onNext（“one”）,onNext（“two”）,onNext（“three”）
*/
Observable.just("one", "two", "three");

from(T[]) / from(Iterable < ? extends T>) : 将传入的数组或 Iterable 拆分成具体对象后，依次发送出来。

/**
* 将依次调用onNext（“one”）,onNext（“two”）,onNext（“three”）
*/
String string[] = new String[]{"one", "two", "three"};
Observable.from(string);

Repeat:创建一个发射特定数据重复多次的observable

repeat重复地发射数据。某些实现允许你重复发射某个数据序列，还有一些允许你限制重复的次数。

RxJava将这个操作符实现为repeat方法。它不是创建一个observable，而是重复发射原始的observable的数据序列，这个序列是无限的，或者通过repeat（n）指定重复次数；

repeat（），repeat（long），repeat（Scheduler）。

empty：创建一个不发射任何数据但是正常终止的observable

never ：创建一个不发射数据也不终止的observable

throw：创建一个不发射数据以一个错误终止的observable

过滤操作

Filter：只发射通过了谓词测试的数据项

observable.just(1,2,3,4).filter(new Func1<Integer,Boolean>() {

@Override
public Boolean call(Integer integer) {
return integer<4;
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(TAG, "call: "+integer);//结果是1，2,3
}
});

SubScribe 订阅

创建了observer和observable之后，再使用subscriber方法将他们连接起来

observable.subscribe(observer)

除了 subscribe(Observer) 和 subscribe(Subscriber) ，subscribe() 还支持不完整定义的回调，RxJava 会自动根据定义创建出 Subscriber 。形式如下：

Action1<String> action1 = new Action1<String>() {
//onNext()
@Override
public void call(String s) {
Log.d(TAG, "call: ");
}
};
Action1<Throwable> throwableAction1 = new Action1<Throwable>() {
//onError()
@Override
public void call(Throwable throwable) {
Log.d(TAG, "call: ");
}
};
Action0 action0 = new Action0() {
//onCompleted
@Override
public void call() {
Log.d(TAG, "call: ");
}
};
observable.subscribe(action1);//自动创建Subscriber，action1 中的call方法来定义onnext;
observable.subscribe(action1, throwableAction1);//自动创建Subscriber，action1 中的call方法来定义onnext,throwableAction1定义onError;
observable.subscribe(action1, throwableAction1, action0);//自动创建Subscriber，action1 中的call方法来定义onnext,
// throwableAction1定义onError,action0定义了oncomplete;
observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscri我be(observer);

在 RxJava 的默认规则中，事件的发出和消费都是在同一个线程的。观察者模式本身的目的就是『后台处理，前台回调』的异步机制，因此异步对于 RxJava 是至关重要的。而要实现异步，则需要用到 RxJava 的另一个概念： Scheduler 。

线程控制Scheduler

在哪个线程调用 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到 Scheduler （调度器）。

Scheduler的API(一）

Scheduler——调度器，相当于线程控制器，RxJava通过它来指定每一段代码应该运行在什么样的线程。

RxJava已经内置了几个scheduler，他们适合大多数使用场景

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

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

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

Schedulers.computation():计算所使用的Schedule。这个计算指的是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()) 的使用方式非常常见，它适用于多数的 『后台线程取数据，主线程显示』的程序策略。

变换

所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。

API

将string 123转换为int型123：

Observable.just("123").map(new Func1<String, Integer>() {
@Override
public Integer call(String s) {
return Integer.getInteger(s);
}
}).subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {

}

@Override
public void onNext(Integer integer) {

}
});

map():事件对象的直接变换，具体功能上面代码已经展示过了。是RxJava最常用的变换。

flatMap()：这是一个很有用但是非常难理解的变换。举个栗子说明：假设有一个数据结构：学生，现在需要打印一组学生的姓名。

Student[] students = ...;
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onNext(String name) {
Log.d(tag, name);
}
...
};
Observable.from(students)
.map(new Func1<Student, String>() {
@Override
public String call(Student student) {
return student.getName();
}
})
.subscribe(subscriber);

如果我们需要打印的是学生报名的课程的话，那么map这个方法就不实用了，因为一个学生可以对应多个课程，所以这时候我们就可以实用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。

扩展：由于可以在嵌套的 Observable 中添加异步代码， flatMap() 也常用于嵌套的异步操作，例如嵌套的网络请求。示例代码（Retrofit + RxJava）：

networkClient.token() // 返回 Observable<String>，在订阅时请求 token，并在响应后发送 token
.flatMap(new Func1<String, Observable<Messages>>() {
@Override
public Observable<Messages> call(String token) {
// 返回 Observable<Messages>，在订阅时请求消息列表，并在响应后发送请求到的消息列表
return networkClient.messages();
}
})
.subscribe(new Action1<Messages>() {
@Override
public void call(Messages messages) {
// 处理显示消息列表
showMessages(messages);
}
});

传统的嵌套请求需要使用嵌套的 Callback 来实现。而通过 flatMap() ，可以把嵌套的请求写在一条链中，从而保持程序逻辑的清晰。

throttleFirst(): 在每次事件触发后的一定时间间隔内丢弃新的事件。常用作去抖动过滤，例如按钮的点击监听器： RxView.clickEvents(button) // RxBinding 代码，后面的文章有解释 .throttleFirst(500, TimeUnit.MILLISECONDS) // 设置防抖间隔为 500ms .subscribe(subscriber); 妈妈再也不怕我的用户手抖点开两个重复的界面啦。

此外， RxJava 还提供很多便捷的方法来实现事件序列的变换，这里就不一一举例了。

变换的原理 lift()

这些变换虽然功能各有不同，但实质上都是针对事件序列的处理和再发送。而在 RxJava 的内部，它们是基于同一个基础的变换方法： lift(Operator)。首先看一下 lift() 的内部实现（仅核心代码）：

// 注意：这不是 lift() 的源码，而是将源码中与性能、兼容性、扩展性有关的代码剔除后的核心代码。
// 如果需要看源码，可以去 RxJava 的 GitHub 仓库下载。
public <R> Observable<R> lift(Operator<? extends R, ? super T> operator) {
return Observable.create(new OnSubscribe<R>() {
@Override
public void call(Subscriber subscriber) {
Subscriber newSubscriber = operator.call(subscriber);
newSubscriber.onStart();
onSubscribe.call(newSubscriber);
}
});
}

在 Observable 执行了 lift(Operator) 方法之后，会返回一个新的 Observable，这个新的 Observable 会像一个代理一样，负责接收原始的 Observable 发出的事件，并在处理后发送给 Subscriber。

线程控制 Scheduler（二）

1) Scheduler 的 API (二)

前面讲到了，可以利用 subscribeOn() 结合 observeOn() 来实现线程控制，让事件的产生和消费发生在不同的线程。可是在了解了 map() flatMap() 等变换方法后，有些好事的（其实就是当初刚接触 RxJava 时的我）就问了：能不能多切换几次线程？

答案是：能。因为 observeOn() 指定的是 Subscriber 的线程，而这个 Subscriber 并不是（严格说应该为『不一定是』，但这里不妨理解为『不是』）subscribe() 参数中的 Subscriber ，而是 observeOn() 执行时的当前 Observable 所对应的 Subscriber ，即它的直接下级 Subscriber 。换句话说，observeOn() 指定的是它之后的操作所在的线程。因此如果有多次切换线程的需求，只要在每个想要切换线程的位置调用一次 observeOn() 即可。上代码：

Observable.just(1, 2, 3, 4) // IO 线程，由 subscribeOn() 指定
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.newThread())
.map(mapOperator) // 新线程，由 observeOn() 指定
.observeOn(Schedulers.io())
.map(mapOperator2) // IO 线程，由 observeOn() 指定
.observeOn(AndroidSchedulers.mainThread)
.subscribe(subscriber);  // Android 主线程，由 observeOn() 指定

如上，通过 observeOn() 的多次调用，程序实现了线程的多次切换。

不过，不同于 observeOn() ， subscribeOn() 的位置放在哪里都可以，但它是只能调用一次的。

又有好事的（其实还是当初的我）问了：如果我非要调用多次 subscribeOn() 呢？会有什么效果？

这个问题先放着，我们还是从 RxJava 线程控制的原理说起吧。

2) Scheduler 的原理（二）

其实， subscribeOn() 和 observeOn() 的内部实现，也是用的 lift()。

不同的是， subscribeOn() 的线程切换发生在 OnSubscribe 中，即在它通知上一级 OnSubscribe 时，这时事件还没有开始发送，因此 subscribeOn() 的线程控制可以从事件发出的开端就造成影响；而 observeOn() 的线程切换则发生在它内建的 Subscriber 中，即发生在它即将给下一级 Subscriber 发送事件时，因此 observeOn() 控制的是它后面的线程。

3) doOnSubscribe()

在前面讲 Subscriber 的时候，提到过 Subscriber 的 onStart() 可以用作流程开始前的初始化。然而 onStart() 由于在 subscribe() 发生时就被调用了，因此不能指定线程，而是只能执行在 subscribe() 被调用时的线程。这就导致如果 onStart() 中含有对线程有要求的代码（例如在界面上显示一个 ProgressBar，这必须在主线程执行），将会有线程非法的风险，因为有时你无法预测 subscribe() 将会在什么线程执行。

而与 Subscriber.onStart() 相对应的，有一个方法 Observable.doOnSubscribe() 。它和 Subscriber.onStart() 同样是在 subscribe() 调用后而且在事件发送前执行，但区别在于它可以指定线程。默认情况下， doOnSubscribe() 执行在 subscribe() 发生的线程；而如果在 doOnSubscribe() 之后有 subscribeOn() 的话，它将执行在离它最近的 subscribeOn() 所指定的线程。

Observable.create(onSubscribe)
.subscribeOn(Schedulers.io())
.doOnSubscribe(new Action0() {
@Override
public void call() {
progressBar.setVisibility(View.VISIBLE); // 需要在主线程执行
}
})
.subscribeOn(AndroidSchedulers.mainThread()) // 指定主线程
.observeOn(AndroidSchedulers.mainThread())
.subscribe(subscriber);

RxJava的实用场景和使用方式

1 与Retrofit的结合

Retrofit 除了提供了传统的 Callback 形式的 API，还有 RxJava 版本的 Observable 形式 API。下面我用对比的方式来介绍 Retrofit 的 RxJava 版 API 和传统版本的区别。

以获取一个 User 对象的接口作为例子。使用Retrofit 的传统 API，你可以用这样的方式来定义请求：

@GET("/user")
public void getUser(@Query("userId") String userId, Callback<User> callback);

在程序的构建过程中， Retrofit 会把自动把方法实现并生成代码，然后开发者就可以利用下面的方法来获取特定用户并处理响应：

getUser(userId, new Callback<User>() {
@Override
public void success(User user) {
userView.setUser(user);
}

@Override
public void failure(RetrofitError error) {
// Error handling
...
}
};

而使用 RxJava 形式的 API，定义同样的请求是这样的：

/**
*
* @param userid
* @return  Retrofit 与 RxJava的嵌套使用
*/

@GET("/users")
public Observable<ApiInfo> getinfo(@Query("userid") String userid);

使用的时候是这样的：

getUser(userId)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<User>() {
@Override
public void onNext(User user) {
userView.setUser(user);
}

@Override
public void onCompleted() {
}

@Override
public void onError(Throwable error) {
// Error handling
...
}
});

当 RxJava 形式的时候，Retrofit 把请求封装进 Observable ，在请求结束后调用 onNext() 或在请求失败后调用 onError()。

对比来看， Callback 形式和 Observable 形式长得不太一样，但本质都差不多，而且在细节上 Observable 形式似乎还比 Callback 形式要差点。那 Retrofit 为什么还要提供 RxJava 的支持呢？

因为它好用啊！从这个例子看不出来是因为这只是最简单的情况。而一旦情景复杂起来， Callback 形式马上就会开始让人头疼。比如：

假设这么一种情况：你的程序取到的 User 并不应该直接显示，而是需要先与数据库中的数据进行比对和修正后再显示。使用 Callback 方式大概可以这么写：

getUser(userId, new Callback<User>() {
@Override
public void success(User user) {
new Thread() {
@Override
public void run() {
processUser(user); // 尝试修正 User 数据
runOnUiThread(new Runnable() { // 切回 UI 线程
@Override
public void run() {
userView.setUser(user);
}
});
}).start();
}

@Override
public void failure(RetrofitError error) {
// Error handling
...
}
};

这时候，如果用 RxJava 的形式，就好办多了。 RxJava 形式的代码是这样的：‘’

getUser(userId)
.doOnNext(new Action1<User>() {
@Override
public void call(User user) {
processUser(user);
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<User>() {
@Override
public void onNext(User user) {
userView.setUser(user);
}

@Override
public void onCompleted() {
}

@Override
public void onError(Throwable error) {
// Error handling
...
}
});

再举一个例子：假设 /user 接口并不能直接访问，而需要填入一个在线获取的 token ，代码应该怎么写？

Callback 方式，可以使用嵌套的 Callback：

@GET("/token")
public Observable<String> getToken();

@GET("/user")
public Observable<User> getUser(@Query("token") String token, @Query("userId") String userId);

...

getToken()
.flatMap(new Func1<String, Observable<User>>() {
@Override
public Observable<User> onNext(String token) {
return getUser(token, userId);
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<User>() {
@Override
public void onNext(User user) {
userView.setUser(user);
}

@Override
public void onCompleted() {
}

@Override
public void onError(Throwable error) {
// Error handling
...
}
});

RxBinding

RxBinding 是 Jake Wharton 的一个开源库，它提供了一套在 Android 平台上的基于 RxJava 的 Binding API。所谓 Binding，就是类似设置 OnClickListener 、设置 TextWatcher 这样的注册绑定对象的 API。

举个设置点击监听的例子。使用 RxBinding ，可以把事件监听用这样的方法来设置：

Button button = ...;
RxView.clickEvents(button) // 以 Observable 形式来反馈点击事件
.subscribe(new Action1<ViewClickEvent>() {
@Override
public void call(ViewClickEvent event) {
// Click handling
}
});

看起来除了形式变了没什么区别，实质上也是这样。甚至如果你看一下它的源码，你会发现它连实现都没什么惊喜：它的内部是直接用一个包裹着的 setOnClickListener() 来实现的。然而，仅仅这一个形式的改变，却恰好就是 RxBinding 的目的：扩展性。通过 RxBinding 把点击监听转换成 Observable 之后，就有了对它进行扩展的可能。扩展的方式有很多，根据需求而定。一个例子是前面提到过的 throttleFirst() ，用于去抖动，也就是消除手抖导致的快速连环点击：

RxView.clickEvents(button)
.throttleFirst(500, TimeUnit.MILLISECONDS)
.subscribe(clickAction);