RxJava是一个基于可观测序列组成的异步的、基于事件的库。通俗一点说就是RxJava它是一个异步库，这个异步库可以让我们用非常简洁的代码来处理复杂数据流或者事件。
RxJava中两个最最基础的概念，一个是Observable，还有一个是Observer，其中Observable我们称之为被观察者，Observer称之为观察者，Observable用户发送消息，而Observer用于消费消息，在实际开发中，我们更多的是选择Observer的一个子类Subscriber来消费消息。在消息发送的过程中，Observable可以发送任意数量任意类型的消息（甚至一个将一个触摸事件当作一个消息发出），当Observable所发送的消息被成功处理或者消息出错时，一个流程结束。Observable会用它的每一个Subscriber（观察者）的onNext方法进行消息处理，在消息成功处理后以onComplete()方法结束流程，如果消息在处理的过程中出现了任何异常，则以onError()方法结束流程。比如下面几行代码：
Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("哈哈哈哈");
subscriber.onNext("lalalala");
subscriber.onCompleted();
}
})
.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
Log.d("google_lenve_fb", "onCompleted: onCompleted()");
}
@Override
public void onError(Throwable e) {
Log.d("google_lenve_fb", "onError:onError() ");
}
@Override
public void onNext(String s) {
Log.d("google_lenve_fb", "onNext: " + s);
}
});
通过调用Observable的create方法来创建一个消息源，在它的onCall方法调用next方法来发送两条消息，当两条消息发送完成之后，调用onComplete方法表示消息发送完毕！subscribe表示订阅一条消息，在订阅的时候我们可以传入一个Observer对象，也可以传入一个Subscriber对象，这两个对象中的方法都是一样的，在onNext方法中处理消息，当消息处理完成之后会自动的调用onComplete方法，如果消息处理过程中出错，则会调用onError方法
如果我在onNext方法执行一行 1/0 ，onNext方法改成下面的样子：
public void onNext(String s) {
Log.d("google_lenve_fb", "onNext: " + s);
int i = 1 / 0;
}
打印结果如下：
onNext: 哈哈哈哈
onError:onError()
当第一条消息打印出来之后，执行1/0时抛异常，直接调用了onError方法，第二条消息将不再处理
常见API
- 1 from函数
Observale中的from函数接受一个数组，这个方法返回一个按数组元素的顺序来发射这些数据的Observale - 2 just函数
just函数它接受最多10个参数，返回一个按参数顺序发射这些数据的Observable - 3 map函数
map函数可以对Observable创建的原始数据进行二次加工，然后再被观察者获取
Observable.from(new String[]{"Observable Test stringOne","Observable Test stringTwo"})
.map(new Func1<String, String>() {
@Override
public String call(String s) {
return s + "---end";
}
})
.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve_fb", "onNext: " + s);
}
});
打印结果如下：
Observable Test stringOne"---end
Observable Test stringTwo"---end
RxJava的使用都是链式编程，使用map函数时数据来源可以各种各样
- 4 flatMap函数
flatMap函数接受一个Observable函数作为输入函数，然后在这个输入的基础上再创建一个新的Observable进行输出，比如下面一段代码：
[java] view plain copy print?Observable.just("Test stringOne", "Test stringTwo")
.flatMap(new Func1<String, Observable<String>>() {
@Override
public Observable<String> call(String s) {
return Observable.from(new String[]{s + "endOne", s + "endTwo"});
}
}).subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve_fb", "onNext: " + s);
}
});
打印结果：
Test stringOneendOne
Test stringOneendTwo
Test stringTwoendOne
Test stringTwoendTwo
- 5 scan函数
scan函数是一个累加器函数，对于Observable发射的每项数据进行累加，并将累加
的结果返回，如下：
[java] view plain copy print?Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
.scan(new Func2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer integer, Integer integer2) {
return integer + integer2;
}
})
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
Log.d("google_lenve_fb", "onNext: " + integer);
}
});
Observable.from(new String[]{"明", "月", "别", "枝", "惊", "鹊"})
.scan(new Func2<String, String, String>() {
@Override
public String call(String s, String s2) {
return s + s2;
}
})
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve", "onNext: " + s);
}
});
打印结果如下：
D/google_lenve_fb: onNext: 1
D/google_lenve_fb: onNext: 3
D/google_lenve_fb: onNext: 6
D/google_lenve_fb: onNext: 10
D/google_lenve_fb: onNext: 15
D/google_lenve_fb: onNext: 21
D/google_lenve_fb: onNext: 28
D/google_lenve_fb: onNext: 36
D/google_lenve_fb: onNext: 45
D/google_lenve_fb: onNext: 55
D/google_lenve: onNext: 明
D/google_lenve: onNext: 明月
D/google_lenve: onNext: 明月别
D/google_lenve: onNext: 明月别枝
D/google_lenve: onNext: 明月别枝惊
D/google_lenve: onNext: 明月别枝惊鹊
-
6 elementAt函数
elementAt函数表示获取数据源中的第N项数据输出 -
7 merge函数
merge函数可以用来合并多个Observable数据源，然后将合并后的数据在一起输出，如下：
[java] view plain copy print?Observable<String> observable1 = Observable.just("十里楼台倚翠薇", "百花深处杜鹃啼");
Observable<String> observable2 = Observable.just("殷勤自与行人语","不似流莺取次飞");
Observable.merge(observable1, observable2)
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve", "onNext: "+s);
}
});
打印结果如下：
D/google_lenve: onNext: 十里楼台倚翠薇
D/google_lenve: onNext: 百花深处杜鹃啼
D/google_lenve: onNext: 殷勤自与行人语
D/google_lenve: onNext: 不似流莺取次飞
- 8 zip函数
zip函数用来合并多个Observable的数据源，但是与merge不同的是，zip函数中可以对数据源进行二次操作，而不是简单的合并，代码如下：
[java] view plain copy print?Observable<String> observable1 = Observable.just("十里楼台倚翠薇", "百花深处杜鹃啼");
Observable<String> observable2 = Observable.just("问君能有几多愁","恰似一江春水向东流");
Observable.zip(observable1, observable2, new Func2<String, String, Object>() {
@Override
public Object call(String s, String s2) {
return s+s2;
}
})
.subscribe(new Subscriber<Object>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Object o) {
Log.d("google_lenve", "onNext: "+o.toString());
}
});
打印结果如下：
D/google_lenve: onNext: 十里楼台倚翠薇问君能有几多愁
D/google_lenve: onNext: 百花深处杜鹃啼恰似一江春水向东流
- 9 过滤函数
Observable函数中还有其他一些好用的过滤函数
private void m7() {
Observable.just(10, 12, 13, 14, 15, 16, 16, 16, 17, 18)
//在数据序列的开头插入一条指定的项1
.startWith(2)
.filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer > 0;
}
})
//只发射前N个元素
.take(2)
//只发射最后N个元素
.takeLast(2)
//只发射第一个元素
.first()
//只发射最后一个元素
.last()
//跳过前两个
.skip(2)
//跳过最后两个
.skipLast(2)
//数据过滤，过滤掉重复数据
.distinct()
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
Log.d("google_lenve_fb", "onNext: " + integer);
}
});
网友评论