Android 响应式编程 RxJava2 - 推酷
Android 响应式编程 RxJava2
1.RxJava2 基础
RxJava可以浓缩为异步两个字，其核心的东西不外乎两个， Observables（被观察者） 和 Observable（观察者）。Observables可以发出一系列的 事件（例如网络请求、复杂计算、数据库操作、文件读取等），事件执行结束后交给Observable 的回调处理。
1.RxJava2 的观察者模式
观察者模式是对象的行为模式，也叫做发布-订阅(Publish/Subscribe)模式、模型-视图(Model/View)模式、源-监听器(Source/Listener)模式或从属者(Dependents)模式。
什么是观察者模式？举个栗子，Android中View的点击监听器的实现，View是被观察者，OnClickListener对象是观察者，Activity要如何知道View被点击了？那就是派一个OnClickListener对象，入驻View，与View达成一个订阅关系，一旦View被点击了，就通过OnClickListener对象的OnClick方法传达给Activity。采用观察者模式可以避免去轮询检查，节约有限的cpu资源。
RxJava 作为一个工具库，使用的便是通用形式的观察者模式：
普通事件：onNext()，相当于 onClick()、onEvent()；特殊事件：onCompleted() 和 onError()
如图所示，RxJava 的基本概念分别为：Observable(被观察者，事件源)，Observer(观察者，订阅者)，subscribe (订阅)、事件；不同的是，RxJava 把多个事件看做一个队列，并对每个事件单独处理。在一个队列中 onCompleted() 和 onError()，只有一个会被调用。如果调用了 onCompleted() 就说明队列执行完毕，没有出现异常，否则调用 onError() 方法并终止队列。
2.RxJava2 响应式编程结构
什么是响应式编程？举个栗子，a = b + 这句代码将b+c的值赋给a，而之后如果b和c的值改变了不会影响到a，然而，对于响应式编程，之后b和c的值的改变也动态影响着a，意味着a会随着b和c的变化而变化。
响应式编程的组成为Observable/Operator/Subscriber，RxJava在响应式编程中的基本流程如下：
这个流程，可以简单的理解为：Observable -& Operator1 -& Operator2 -& Operator3 -& Subscriber
1. Observable发出一系列事件，他是事件的产生者；
2. Subscriber负责处理事件，他是事件的消费者；
3. Operator是对Observable发出的事件进行修改和变换；
4. 若事件从产生到消费不需要其他处理，则可以省略掉中间的Operator，从而流程变为Obsevable -& Subscriber；
5. Subscriber通常在主线程执行，所以原则上不要去处理太多的事务，而这些复杂的处理则交给Operator；
3.创建一个完整的 RxJava2 调用
首先需要添加 RxJava2 在 Android 中的 Gradle 依赖：
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile &io.reactivex.rxjava2:rxjava:2.0.8&
RxJava2 可以通过下面这几种方法创建被观察者：
// 发送对应的方法
Observable.create(new ObservableOnSubscribe
// 默认在主线程里执行该方法
public void subscribe(@NonNull ObservableEmitte
e) throws Exception {
e.onNext(&Hello&);
e.onNext(&World&);
// 结束标识
e.onComplete();
// 发送多个数据
Observable.just(&Hello&, &World&);
// 发送数组
Observable.fromArray(&Hello&, &World&);
// 发送一个数据
Observable.fromCallable(new Callable
public String call() throws Exception {
return &Hello&;
RxJava2 支持链式编程，下来我们创建被观察者，然后创建观察者并订阅：
// 创建被观察者
Observable.just(&Hello&, &World&)
// 将被观察者切换到子线程
.subscribeOn(Schedulers.io())
// 将观察者切换到主线程
.observeOn(AndroidSchedulers.mainThread())
// 创建观察者并订阅
.subscribe(new Observer&String&() {
// Disposable 相当于RxJava1.x中的
Subscription，
用于解除订阅
public void onSubscribe(Disposable d) {
disposable =
public void onNext(String s) {
Log.i(&JAVA&, &被观察者向观察者
发送的数据:& + s);
if (s == &-1&) {
// &-1& 时为异常
数据，解除订阅
disposable.dispose();
public void onError(Throwable e) {
public void onComplete() {
一旦 Observer 订阅了 Observable，Observable 就会调用 Observer 的 onNext()、onCompleted()、onError() 等方法。至此一个完整的 RxJava 调用就完成了。看一下输出的Log：
I/JAVA: 被观察者向观察者发送的数据:Hello I/JAVA: 被观察者向观察者发送的数据:World
若喜欢简洁、定制服务，那么可以实现的方法跟上面的实现方法是对应起来的，大家看参数就知道哪个对应哪个了，你可以通过new Consumer（不需要实现的方法你可以不写，看上去更简洁），Consumer就是消费者的意思，可以理解为消费了 onNext 等事件：
Observable.just(&Hello&, &World&)
.subscribe(new Consumer&String&() {
public void accept(@NonNull String s) throws Exception {
Log.i(&JAVA&, &被观察者向观察者发送的数据:& + s);
}, new Consumer&Throwable&() {
public void accept(@NonNull Throwable throwable) throws Exception {
}, new Action() {
public void run() throws Exception {
}, new Consumer&Disposable&() {
public void accept(@NonNull Disposable disposable ) throws Exception {
4.RxJava2 的操作符
RxJava中提供了大量不同种类，不同场景的Operators(操作符)，RxJava的强大性就来自于它所定义的操作符。主要分类：
其中有一些高频使用的操作符如下：
5.RxJava2 线程调度器
调度器 Scheduler 用于控制操作符和被观察者事件所执行的线程，不同的调度器对应不同的线程。RxJava提供了5种调度器：
可以使用 subscribeOn() 和 ObserveOn() 操作符进行线程调度，让 Observable 在一个特定的调度器上执行。subscribeOn() 指定 subscribe() 所发生的线程，事件产生的线程。ObserveOn() 指定 Observer 所运行在的线程，事件消费的线程。
6.RxJava2 模拟发送验证码倒计时功能
public void onCodeClick() {
final long count = 60; // 设置60秒
Observable.interval(0, 1, TimeUnit.SECONDS)
.take(count + 1)
.map(new Function&Long, Long&() {
public Long apply(@NonNull Long aLong) throws Exception {
return count - aL // 由于是倒计时，需要将倒计时的数字反过来
.observeOn(AndroidSchedulers.mainThread())
.doOnSubscribe(new Consumer&Disposable&() {
public void accept(@NonNull Disposable disposable) throws Exception {
button.setEnabled(false);
button.setTextColor(Color.GRAY);
.subscribe(new Observer&Long&() {
public void onSubscribe(Disposable d) {
public void onNext(Long aLong) {
button.setText(aLong + &秒后重发&);
public void onError(Throwable e) {
public void onComplete() {
button.setEnabled(true);
button.setTextColor(Color.RED);
button.setText(&发送验证码&);
2.RxJava2 系列框架
3.RxJava2 与 Retrofit 的使用
RxJava 与 Retrofit 的使用，更像我们的 AsyncTask，通过网络获取数据然后通过 Handler 更新UI。首先需要导\入依赖：
compile 'com.squareup.retrofit2:retrofi t:2.2.0''
compile 'com.squareup.retrofit2:converte
r-gson:2.2.0'
compile 'com.squareup.retrofit2:adapter-rxjava
1.模拟用户登陆获取用户数据
1.Bean对象：
public class UserParam {
private String param1;
private String param2;
public UserParam(String param1, String param2) {
this.param1 = param1;
this.param2 = param2;
// 省略了 getter setter
public class NetBean {
private FormB
// 省略了 getter setter
public static class FormBean {
// 省略了 getter setter
public class UserBean {
public UserBean(String username, String password) {
this.username =
this.password =
// 省略了 getter setter
2.ApiService，这里返回Observable对象，也就是我们RxJava的被观察者
public interface ApiService {
@FormUrlEncoded
@POST(&/post&)
Observable&NetBean& getUserInfo
(@Field(&username&)String username,
@Field(&password&)String password);
3.RxJava + Retrofit 的实现
// 构建Retrofit
ApiService apiService = new Retrofit.Builder()
.baseUrl(&http://httpbin.org/&)
.addConverterFactory(GsonConverterFactory.create()) // RxJava2与Gson混用
.addCallAdapterFactory(RxJava2CallAdapterFactory .create()) // RxJava2与Retrofit混用
// 构建RxJava
UserParam param = new UserParam(&zhangsan&, &123&);
// 发送param参数
Observable.just(param)
// flatMap方法是用于数据格式转换的方法，参数一表示原数据，
// 参数二表示转换的数据，那么就是通过发送网络参数
，转换成网络返回的数据，调用Retrofit
.flatMap(new Function&UserParam, ObservableSource
&NetBean&&() {
public ObservableSource&NetBean& apply(@NonNull
UserParam userParam)
throws Exception {
// 1.发送网络请求，获取NetBean
return apiService.getUserInfo(userParam.getParam1(), userParam.getParam2());
.flatMap(new Function&NetBean, ObservableSourc
e&UserBean&&() {
public ObservableSource&UserBean&
apply(@NonNullNetBean netBean)throws
Exception {UserBean user = new UserBean(netBean.getForm().getUsername(),
netBean.getForm().getPassword());
// 2.转换NetBean数据为我们需要的UserBean数据
return Observable.just(user);
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer&UserBean&() {
public void accept(@NonNull UserBean userBean)
throws Exception {
Log.i(&JAVA&, && + &用户名:& + userBean.getUsername()
+ &, 密码:& + userBean.getPassword());
2.模拟合并本地与服务器购物车列表
这个案例其实就是用户添加购物车的时候，首先会在本地存储一份，然后发现如果没有网络，那么没办法提交到服务器上，只能等下一次有网络的时候采用本地数据库和服务器数据的合并来实现上传到服务器。
首先需要准备 Retrofit 对象和获取本地数据、网络数据的方法：
private ApiService apiS
protected void onCreate(@Nullable Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
// 构建Retrofit
apiService = new Retrofit.Builder()
.baseUrl(&http://httpbin.org/&)
.addConverterFactory(GsonConverterFactory. create()) // RxJava2与Gson混用
.addCallAdapterFactory(RxJava2CallAdapter Factory.create()) // RxJava2与Retrofit混用
.create(ApiService.class);
* 获取本地数据
private Observable&List&String&& getDataForLocal() {
List&String& list = new ArrayList&&();
list.add(&购物车的商品1&);
list.add(&购物车的商品2&);
return Observable.just(list);
* 获取网络数据
private Observable&List&String&& getDataForNet() {
return Observable.just(&shopName&)
// flatMap方法是用于数据格式转换的方法， 参数一表示原数据，
// 参数二表示转换的数据，那么就是通过发 送网络参数，转换成网络返回的数据，调用Retrofit
.flatMap(new Function&String, Observable Source&NetBean&&() {
public ObservableSource&NetBean& apply (@NonNull String s) throws Exception {
// 1.发送网络请求，获取数据
return apiService.getCartList(s);
}).flatMap(new Function&NetBean, Observable Source&List&String&&&() {
public ObservableSource&List&String&& apply(@NonNull NetBean netBean) throws Exception {
// String shop = netBean.get_$Args257() .getShopName();
String shop = &购物车的商品3&;
List&String& list = new ArrayList&&();
list.add(shop);
// 2.转换NetBean数据为我们需要的List &String&数据
return Observable.just(list);
}).subscribeOn(Schedulers.io());
然后我们就可以创建被观察者并订阅了，来完成合并本地与服务器购物车列表操作：
// merge操作符: 将两个ObservableSource
合并为一个ObservableSource
Observable.merge(getDataForLocal() , getDataForNet())
.subscribe(new Observer &List&String&&() {
public void onSubscribe (Disposable d) {
public void onNext(List &String& strings) {
for (String str: strings) { Log.i(&JAVA&, str); }
public void onError (Throwable e) {
public void onComplete() {
Log.i(&JAVA&, &onComplete&);
最后的打印结果是：
I/JAVA: 购物车的商品1
I/JAVA: 购物车的商品2
I/JAVA: 购物车的商品3
I/JAVA: onComplete
4.RxJava2 与 RxBinding 的使用
1.优化搜索请求
当我们在 EditText 打字时搜索的时候，可能用户会打字很会快，那么我们就没有必要一直发送网络请求，请求搜索结果，我们可以通过当用户打字停止后的延时500毫秒再发送搜索请求：
// RxTextView.textChanges(edittext):
Rxbinding用法
RxTextView.textChanges(editText)
// 表示延时多少秒后执行， 当你敲完字之后停下来的半秒就会执行下面语句
.debounce(500, TimeUnit.MILLISECONDS)
// 数据转换 flatMap: 当同时多个数据请 求访问的时候，前面的网络数据会覆盖后面的网络 数据 // 数据转换 switchMap: 当同时多个网络请 求访问的时候，会以最后一个发送请求为准，前面网络 数据会被最后一个覆盖
.switchMap(new Function&CharSequence, ObservableSource&List&String&&&() {
public ObservableSource&List&String&& apply(
@NonNull CharSequence charSequence) throws Exception {
// 网络请求操作，获取我们需要的数据
List&String& list = new ArrayList &String&();
list.add(&2017&);
list.add(&2018&);
return Observable.just(list);
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer&List&String&&() {
public void accept(@NonNull List&String& strings) throws Exception {
Log.i(&JAVA&, strings.toString());
2.优化点击请求
当用户一直点击一个按钮的时候，我们不应该一直调用访问网络请求，而是 1秒内，只执行一次网络请求。
RxView.clicks(button).throttleFirst
(1, TimeUnit.SECONDS)
.subscribe(new Observer&Object&() {
public void onSubscribe(Disposable d) {
public void onNext(Object o) {
Log.i(&JAVA&, &onClick&);
public void onError(Throwable e) {
public void onComplete() {
5.RxJava2 踩过的一些坑
1.未解除订阅而引起的内存泄漏
举个例子，对于前面常用操作符 interval 做周期性操作的例子，并没有使之停下来的，没有去控制订阅的生命周期，这样，就有可能引发内存泄漏。所以，在 Activity 的 onDestroy() 方法执行的时候或者不需要继续执行的时候应该解除订阅。
已发表评论数()
请填写推刊名
描述不能大于100个字符!
权限设置： 公开
仅自己可见
正文不准确
标题不准确
排版有问题
主题不准确
没有分页内容
图片无法显示
视频无法显示
与原文不一致您所在位置： &
&nbsp&&nbsp&nbsp&&nbsp
RxJava 常用操作符.doc 17页
本文档一共被下载：
次 ,您可全文免费在线阅读后下载本文档。
下载提示
1.本站不保证该用户上传的文档完整性，不预览、不比对内容而直接下载产生的反悔问题本站不予受理。
2.该文档所得收入(下载+内容+预览三)归上传者、原创者。
3.登录后可充值，立即自动返金币，充值渠道很便利
需要金币：350 &&
RxJava 常用操作符
你可能关注的文档：
··········
RxJava常用操作符1Observable的创建1.1from()转换集合为一个每次发射集合中一个元素的Observable对象。可用来遍历集合。方法列表：publicstatic&T&Observable&T&from(Future&?extendsT&future)publicstatic&T&Observable&T&from(Future&?extendsT&future,longtimeout,TimeUnitunit)publicstatic&T&Observable&T&from(Future&?extendsT&future,Schedulerscheduler)publicstatic&T&Observable&T&from(Iterable&?extendsT&iterable)publicstatic&T&Observable&T&from(T[]array)栗子：//1.遍历集合Observable&String&observable=Observable.from(newString[]{&hello&,&hi&});12//2.使用Future创建Observable，Future表示一个异步计算的结果。FutureTask&String&futureTask=newFutureTask&String&(newCallable&String&(){@OverridepublicStringcall()throwsException{//TODO执行异步操作并返回数据return&hihi&;}});Scheduler.Workerworker=Schedulers.io().createWorker();worker.schedule(newAction0(){@Overridepublicvoidcall(){futureTask.run();}});Observable&String&observable=Observable.from(futureTask);1.2just()转换一个或多个Object为依次发射这些Object的Observable对象。方法列表：publicstatic&T&Observable&T&just(finalTvalue)publicstatic&T&Observable&T&just(Tt1,Tt2)publicstatic&T&Observable&T&just(Tt1,Tt2,Tt3)publicstatic&T&Observable&T&just(Tt1,Tt2,Tt3,Tt4)publicstatic&T&Observable&T&just(Tt1,Tt2,Tt3,Tt4,Tt5)publicstatic&T&Observable&T&just(Tt1,Tt2,Tt3,Tt4,Tt5,Tt6)publicstatic&T&Observable&T&just(Tt1,Tt2,Tt3,Tt4,Tt5,Tt6,Tt7)publicstatic&T&Observable&T&just(Tt1,Tt2,Tt3,Tt4,Tt5,Tt6,Tt7,Tt8)publicstatic&T&Observable&T&just(Tt1,Tt2,Tt3,Tt4,Tt5,Tt6,Tt7,Tt8,Tt9)publicstatic&T&Observable&T&just(Tt1,Tt2,Tt3,Tt4,Tt5,Tt6,Tt7,Tt8,Tt9,Tt10)栗子：Observable&String&observable=Observable.just(&hello&);//使用just()遍历几个元素Observable&String&observable=Observable.just(&hello&,&hi&,&...&);//使用from()方法遍历，效果和just()一样。String[]stringArrs=newString[]{&hello&q
正在加载中，请稍后...使用RxJava实现android里面的倒计时效果
学习RxJava已经很久了，一直没有在项目中使用，之前项目需求比较多，一直没有时间去做一些示例，最近抽空复习了下之前的知识，算是对自己知识的一个笔记吧！
使用之前记得加上依赖哦！
compile 'io.reactivex:rxjava:1.2.6'
compile 'io.reactivex:rxandroid:1.2.1'
* 使用RxJava实现倒计时
private void countDown() {
final long count = 30;
Observable.interval(1, TimeUnit.SECONDS)
.map(new Func1&Long, Long&() {
public Long call(Long integer) {
return count -
.doOnSubscribe(new Action0() {
public void call() {
cutdown.setEnabled(false);
cutdown.setBackgroundColor(Color.DKGRAY);
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer&Long&() {
public void onCompleted() {
cutdown.setEnabled(true);
cutdown.setBackgroundColor(getResources().getColor(R.color.colorPrimary));
cutdown.setText("倒计时");
public void onError(Throwable e) {
public void onNext(Long aLong) {
cutdown.setText("倒计时" + aLong);
&&相关文章推荐
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：750次
排名：千里之外rx_mvp_..（2）
RxJava 操作符
阅读本文前请先了解
的基本使用。转自 http://blog.csdn.net/u/article/details/
参考文档：
1.1 from( )
转换集合为一个每次发射集合中一个元素的
对象。可用来遍历集合。
方法列表：
Observable&String& observable = Observable.from(new String[]{&hello&, &hi&});1212
FutureTask&String& futureTask = new FutureTask&String&(new Callable&String&() {
public String call() throws Exception {
return &hihi&;
Scheduler.Worker worker = Schedulers.io().createWorker();
worker.schedule(new Action0() {
public void call() {
futureTask.run();
Observable&String& observable = Observable.from(futureTask);123456789101112131415161718123456789101112131415161718
1.2 just( )
转换一个或多个 Object 为依次发射这些 Object 的
方法列表：
Observable&String& observable = Observable.just(&hello&);
Observable&String& observable = Observable.just(&hello&, &hi&, &...&);
String[] stringArrs = new String[]{&hello&, &hi&, &...&};
Observable&String& observable = Observable.from(stringArrs);1234567812345678
just() 方法可传入 1~10 个参数，也就说当元素个数小于等于 10 的时候既可以使用 just() 也可以使用from()，否则只能用
from() 方法。
1.3 create( )
返回一个在被 OnSubscribe 订阅时执行特定方法的
方法列表：
Observable.OnSubscribe&String& onSubscribe = new Observable.OnSubscribe& String &() {
public void call(Subscriber&? super String & subscriber) {
subscribe.onNext(&hello&);
subscribe.onCompleted();
Observable&Object& observable = Observable.create(onSubscribe);123456789123456789
此方法不常用，大多数时候都是使用 just( )、form( ) 等方法，如上面那串代码就可以写成：
Observable&Object& observable = Observable.just(&hello&);11
1.4 interval( )
返回一个每隔指定的时间间隔就发射一个序列号的
对象，可用来做倒计时等操作。
方法列表：
// 每隔 1 s 发送一个序列号，序列号从 0 开始，每次累加 1。
Observable&Long& observable = Observable.interval(1, TimeUnit.SECONDS)1212
创建一个在指定延迟时间后发射一条数据（ 固定值：0 ）的
对象，可用来做定时操作。
方法列表：
// 定时 3 s
Observable&Long& observable = Observable.timer(3, TimeUnit.SECONDS)1212
1.6 range( )
创建一个发射指定范围内的连续整数的
方法列表：
// 依次发射 5、6、7
Observable&Integer& observable = Observable.range(5, 3);1212
1.7 empty()
创建一个不发射任何数据就发出 onCompleted() 通知的
方法列表：
Observable&Object& observable = Observable.empty();1212
1.8 error( )
创建不发射任何数据就发出 onError 通知的
方法列表：
Observable&Object& observable = Observable.error(new Throwable(&message&));1212
1.9 never()
创建一个不发射任何数据和通知的
方法列表：
Observable&Object& observable = Observable.never();11
1.10 defer( )
在订阅的时候才会创建 Observable 对象；每一次订阅都创建一个新的
方法列表：
Observable&String& observable = Observable.defer(new Func0&Observable&String&&() {
public Observable&String& call() {
return Observable.just(&string&);
});123456123456
2.1 repeat( )
使Observable 对象在发出 onNext() 通知之后重复发射数据。重做结束才会发出 onComplete() 通知，若重做过程中出现异常则会中断并发出onError() 通知。
方法列表：
Observable&String& observable = Observable.just(&string&);
observable.repeat();
observable.repeat(5);1234512345
2.2 repeatWhen( )
使Observable 对象在发出 onNext() 通知之后有条件的重复发射数据。重做结束才会发出 onCompleted() 通知，若重做过程中出现异常则会中断并发出onError() 通知。
方法列表：
observable.repeatWhen(new Func1&Observable&? extends Void&, Observable&?&&() {
public Observable&?& call(Observable&? extends Void& observable) {
return observable.zipWith(Observable.range(1, 3), new Func2&Void, Integer, Integer&() {
public Integer call(Void aVoid, Integer integer) {
return integer;
}).flatMap(integer -& Observable.timer(1, TimeUnit.SECONDS));
});123456789101112123456789101112
3.1 retry( )
在执行 对象的序列出现异常时，不直接发出 onError() 通知，而是重新订阅该 对象，直到重做过程中未出现异常，则会发出 onNext() 和 onCompleted() 通知；若重做过程中也出现异常，则会继续重试，直到达到重试次数上限，超出次数后发出最新的onError() 通知。
方法列表：
Observable&Integer& observable = Observable.create(new Observable.OnSubscribe&Integer&() {
public void call(Subscriber&? super Integer& subscriber) {
System.out.println(&.......&);
int a = 1 / 0;
subscriber.onNext(a);
subscriber.onCompleted();
observable.retry();
observable.retry(3);
observable.retry(new Func2&Integer, Throwable, Boolean&() {
public Boolean call(Integer integer, Throwable throwable) {
return false;
});1234567891011121314151617181920212212345678910111213141516171819202122
3.2 retryWhen( )
作用： 有条件的执行重试。
方法列表：
observable.retryWhen(new Func1&Observable&? extends Throwable&, Observable&?&&() {
public Observable&?& call(Observable&? extends Throwable& observable) {
return observable.zipWith(Observable.range(1, 3), new Func2&Throwable, Integer, Object&() {
public Object call(Throwable throwable, Integer integer) {
}).flatMap(new Func1&Object, Observable&?&&() {
public Observable&?& call(Object o) {
return Observable.timer(1, TimeUnit.SECONDS);
});12345678910111213141516171234567891011121314151617
4.1 map( )
发射的元素应用于指定的函数，并发送该函数的结果。
方法列表：
Observable.just(2)
.map(new Func1&Integer, String&() {
public String call(Integer integer) {
return String.valueOf(String.format(&原始数据的两倍为: %s&, integer * 2));
});12345671234567
4.2 flatMap( )
对象为另一个
方法列表：
Observable.just(2)
.flatMap(new Func1&Integer, Observable&Long&&() {
public Observable&Long& call(Integer integer) {
return Observable.timer(integer, TimeUnit.SECONDS);
});1234567812345678
5.1 filter( )
只发射满足指定谓词的元素。
方法列表：
Observable.just(-1, -2, 0, 1, 2)
.filter(new Func1&Integer, Boolean&() {
public Boolean call(Integer integer) {
return integer & 0;
});12345671234567
5.2 first( )
返回一个仅仅发射源
发射的第一个［满足指定谓词的］元素的 ，如果如果源
为空，则会抛出一个 NoSuchElementException。
方法列表：
Observable.just(-1, -2, 0, 1, 2).first();
Observable.just(-1, -2, 0, 1, 2)
.first(new Func1&Integer, Boolean&() {
public Boolean call(Integer integer) {
return integer & 0;
Observable.empty().first();12345678910111213141234567891011121314
5.3 last( )
返回一个仅仅发射源
发射的倒数第一个［满足指定谓词的］元素的 ，如果如果源
为空，则会抛出一个 NoSuchElementException。
方法列表：
Observable.just(-1, -2, 0, 1, 2).first();
Observable.just(-1, -2, 0, 1, 2)
.first(new Func1&Integer, Boolean&() {
public Boolean call(Integer integer) {
return integer & 0;
Observable.empty().last();12345678910111213141234567891011121314
5.4 skip( )
跳过前面指定数量或指定时间内的元素，只发射后面的元素。
方法列表：
Observable.just(-1, -2, 0, 1, 2)
.skip(2) // 跳过前两条数据1212
5.5 skipLast( )
跳过前面指定数量或指定时间内的元素，只发射后面的元素。指定时间时会延迟源
发射的任何数据。
方法列表：
Observable.just(-1, -2, 0, 1, 2)
.skip(2) // 跳过后两条数据1212
5.6 take( )
只发射前面指定数量或指定时间内的元素。
方法列表：
Observable.just(-1, -2, 0, 1, 2).take(3)11
5.7 takeLast( )
只发射后面指定数量或指定时间内的元素。指定时间时会延迟源
发射的任何数据。
方法列表：
Observable.just(-1, -2, 0, 1, 2).takeLast(3)11
5.8 sample( )
发射的最后一条数据。
方法列表：
Observable.interval(300, TimeUnit.MILLISECONDS)
.sample(2, TimeUnit.SECONDS)1212
5.9 elementAt( )
只发射指定索引的元素。
方法列表：
Observable.just(-1, -2, 0, 1, 2).elementAt(2)11
5.10 elementAtOrDefault( )
只发射指定索引的元素，若该索引对应的元素不存在，则发射默认值。
方法列表：
Observable.just(-1, -2, 0, 1, 2).elementAtOrDefault(9, -5)11
5.11 ignoreElements( )
不发射任何数据，直接发出 onCompleted() 通知。
方法列表：
Observable.just(-1, -2, 0, 1, 2).ignoreElements()11
5.12 distinct( )
过滤重复的元素，过滤规则是：只允许还没有发射过的元素通过。
方法列表：
Observable.just(-1, -2, 0, 1, 2, 1).distinct();
Observable.just(-1, -2, 0, 1, 2, 1).distinct(new Func1&Integer, Integer&() {
public Integer call(Integer integer) {
return integer * (int)(Math.random() * 10);
});12345678910111234567891011
5.13 debounce( )
每产生结果后，如果在规定的间隔时间内没有产生新的结果，则发射这个结果，否则会忽略这个结果。该操作符会过滤掉发射速率过快的数据项。
方法列表：
Observable&Integer& observable = Observable.create(new Observable.OnSubscribe&Integer&() {
public void call(Subscriber&? super Integer& subscriber) {
for (int i = 1; i & 10; i++) {
subscriber.onNext(i);
Thread.sleep(i * 100);
subscriber.onCompleted();
} catch (Exception e) {
subscriber.onError(e);
observable.debounce(400, TimeUnit.MILLISECONDS)
1234567891011121314151612345678910111213141516
该栗子产生结果为：依次打印5、6、7、8。
附：功能实现
Observable&Integer& traverseObservable = Observable.just(3, 4, 5, 6);
Observable&Long& intervalObservable = Observable.interval(1, TimeUnit.SECONDS);
Func2&Long, Integer, Integer& func2 = new Func2&Long, Integer, Integer&() {
public Integer call(Long aLong, Integer integer) {
return integer;
intervalObservable.zipWith(traverseObservable, func2)
.toBlocking()
.subscribe(new Subscriber&Integer&() {
public void onCompleted() {
System.out.println(&onCompleted&);
public void onError(Throwable e) {
e.printStackTrace();
public void onNext(Integer integer) {
System.out.println(integer);
});123456789101112131415161718192021222324252627282930123456789101112131415161718192021222324252627282930
int startTime = 10;
Observable.interval(0, 1, TimeUnit.SECONDS)
.take(startTime + 1)
.map(new Func1&Long, Long&() {
public Long call(Long time) {
return startTime -
.toBlocking()
.subscribe(new Subscriber&Long&() {
public void onCompleted() {
System.out.println(&倒计时结束&);
public void onError(Throwable e) {
System.out.println(&倒计时出现异常&);
e.printStackTrace();
public void onNext(Long aLong) {
System.out.println(String.format(&倒计时: %s s&, aLong));
});12345678910111213141516171819202122232425262728291234567891011121314151617181920212223242526272829
&&相关文章推荐
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：11362次
排名：千里之外
原创：39篇
转载：57篇
(6)(10)(17)(25)(15)(9)(13)(3)}