RXJava学习笔记（1）

说明

RXJava最近火的不行，各种衍生品已经越来越多的充斥我们的眼球。简单来说，RXJava就是被观察者(Observable)和观察者(Observer)中间通过一种订阅关系(subscribe)来完成事件驱动功能

RXJava 的观察者模式：Observer作为观察者，他提供了onNext、onCompleted、onError三个接口方法，让开发者自行处理从Observable通过subscribe来的回调结果



本系列主要是介绍RXJava的操作符的使用，原理啥的具体自行看大神们的博文

环境搭建

在build.gradle下添加rxjava和rxandroid的库

compile ‘io.reactivex:rxjava:1.1.1’

compile ‘io.reactivex:rxandroid:1.1.0’

创建操作

作为系列的第一篇，我们就从“创建”操作符系列开始讲起

Create操作符

Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
//在使用过程中，最好加上isUnsubscribed判断，以便在没有观察者的时候停止没必要的逻辑处理
if (!subscriber.isUnsubscribed()) {
try {
//分3次分别调用onNext
for (int i=0;i<3;i++) {
subscriber.onNext(i);
}
subscriber.onCompleted();
} catch (Exception e) {
subscriber.onError(e);
}
}
}
}).subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.d("SampleCreateActivity", "成功");
}

@Override
public void onError(Throwable e) {
e.printStackTrace();
}

@Override
public void onNext(Integer integer) {
Log.d("SampleCreateActivity", "integer:" + integer);
}
});

运行结果

com.renyu.rxdemo D/SampleCreateActivity: integer:0
com.renyu.rxdemo D/SampleCreateActivity: integer:1
com.renyu.rxdemo D/SampleCreateActivity: integer:2
com.renyu.rxdemo D/SampleCreateActivity: 成功

在开篇说明的基础上用代码再次进行描述，值得一提的是

Subscriber作为抽象类实现了Observer并且新增了一个onStart方法，便于开发者在subscribe发生之前进行一些操作

Just操作符

Observable.just("string4", "string5", "string6").subscribe(
//onNext
new Action1<String>() {
@Override
public void call(String s) {
Log.d("SampleCreateActivity", s);
}
},
//onError
new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
throwable.printStackTrace();
}
},
//onCompleted
new Action0() {
@Override
public void call() {
Log.d("SampleCreateActivity", "成功");
}
});

运行结果

com.renyu.rxdemo D/SampleCreateActivity: string4
com.renyu.rxdemo D/SampleCreateActivity: string5
com.renyu.rxdemo D/SampleCreateActivity: string6
com.renyu.rxdemo D/SampleCreateActivity: 成功

将其他类型对象或者数据类型（最多10个）转化为Observable对象然后发射出去，值得一提的是这里

action0代表没有入参的回调方法，action1代表有一个入参的回调方法。此处的subscribe提供了多种回调组合，开发者根据需求自行使用

From操作符

String[] stringsArray={"string1", "string2", "string3"};
Observable.from(stringsArray).subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.d("SampleCreateActivity", s);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
throwable.printStackTrace();
}
}, new Action0() {
@Override
public void call() {
Log.d("SampleCreateActivity", "成功");
}
});

运行结果

com.renyu.rxdemo D/SampleCreateActivity: string1
com.renyu.rxdemo D/SampleCreateActivity: string2
com.renyu.rxdemo D/SampleCreateActivity: string3
com.renyu.rxdemo D/SampleCreateActivity: 成功

from与just功能雷同，但是区别在于，

from为一次性发送多个对象，而just只发送一个对象

，例如本例中的stringsArray，如果使用just，则在call中直接返回stringsArray对象回来

Defer操作符

int tempValue=10;
Observable observableJust=Observable.just(tempValue);
Observable observableDefer=Observable.defer(new Func0<Observable<Integer>>() {
@Override
public Observable<Integer> call() {
return Observable.just(tempValue);
}
});
tempValue=11;
observableJust.subscribe(new Action1() {
@Override
public void call(Object o) {
Log.d("SampleCreateActivity", "observableJust  "+o.toString());
}
});
observableDefer.subscribe(new Action1() {
@Override
public void call(Object o) {
Log.d("SampleCreateActivity", "observableDefer  "+o.toString());
}
});

运行结果

com.renyu.rxdemo D/SampleCreateActivity: observableJust  10
com.renyu.rxdemo D/SampleCreateActivity: observableDefer  11

看到区别了吧，defer的区别在于

just是在创建Observable时就赋值完成了，而defer是在订阅之后才赋值完成

interval操作符

subscriptionInterval=Observable.interval(3, 3, TimeUnit.SECONDS).subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
Log.d("SampleCreateActivity", "interval");
}
});

运行结果

02-15 10:45:46.913 14297-14297/com.renyu.rxdemo I/Timeline: Timeline: Activity_idle id: android.os.BinderProxy@42d15eb0 time:21966457
02-15 10:45:49.703 14297-14326/com.renyu.rxdemo D/SampleCreateActivity: interval
02-15 10:45:52.703 14297-14326/com.renyu.rxdemo D/SampleCreateActivity: interval
02-15 10:45:55.703 14297-14326/com.renyu.rxdemo D/SampleCreateActivity: interval
02-15 10:45:58.703 14297-14326/com.renyu.rxdemo D/SampleCreateActivity: interval
02-15 10:46:01.703 14297-14326/com.renyu.rxdemo D/SampleCreateActivity: interval

interval是从0开始，每隔一段时间发送一个数字，起到类似定时器的作用。值得注意的地方是，

如果要显式的显示结果，需要在主线程中订阅

。注意打印的log，本段代码意思是3秒钟之后每隔3秒触发一次，注意入参的意思

timer操作符

Observable.timer(3, TimeUnit.SECONDS).subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
Log.d("SampleCreateActivity", "time");
}
});

运行结果

com.renyu.rxdemo D/SampleCreateActivity: time

延时运行，不多说

range操作符

Observable.range(3, 10).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d("SampleCreateActivity", "integer:" + integer);
}
});

运行结果

com.renyu.rxdemo D/SampleCreateActivity: integer:3
com.renyu.rxdemo D/SampleCreateActivity: integer:4
com.renyu.rxdemo D/SampleCreateActivity: integer:5
com.renyu.rxdemo D/SampleCreateActivity: integer:6
com.renyu.rxdemo D/SampleCreateActivity: integer:7
com.renyu.rxdemo D/SampleCreateActivity: integer:8
com.renyu.rxdemo D/SampleCreateActivity: integer:9
com.renyu.rxdemo D/SampleCreateActivity: integer:10
com.renyu.rxdemo D/SampleCreateActivity: integer:11
com.renyu.rxdemo D/SampleCreateActivity: integer:12

类似于for(int i=3;i<(3+10);i++)

Repeat操作符

Observable.range(3, 2).repeat(2).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d("SampleCreateActivity", "integer:" + integer);
}
});

运行结果

com.renyu.rxdemo D/SampleCreateActivity: integer:3
com.renyu.rxdemo D/SampleCreateActivity: integer:4
com.renyu.rxdemo D/SampleCreateActivity: integer:3
com.renyu.rxdemo D/SampleCreateActivity: integer:4

重复执行

还有一个

repeatWhen

Observable.range(2, 2).repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
@Override
public Observable<?> call(Observable<? extends Void> observable) {
return observable.zipWith(Observable.range(1, 3), new Func2<Void, Integer, Integer>() {
@Override
public Integer call(Void aVoid, Integer integer) {
return integer;
}
}).flatMap(new Func1<Integer, Observable<?>>() {
@Override
public Observable<?> call(Integer integer) {
return Observable.timer(integer, TimeUnit.SECONDS);
}
});
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d("SampleCreateActivity", "integer:" + integer);
}
});

在repeat的基础上，添加条件，实现有条件的重复执行

主要参考文章

给 Android 开发者的 RxJava 详解 扔物线大神的文章，我的RXJava启蒙文章

木水川的博客 对RXJava操作符做了很详细的说明以及提供详细的示例

ReactiveX文档中文翻译 mcxiaoke翻译的RXJava中文文档

Android RxJava使用介绍 job_hesc对RXJava操作符介绍以及简单的示例