前言
RxJava的学习笔记（一）基础概念
RxJava的学习笔记（二）操作符
RxJava的学习笔记（三）线程调度
去年开始在上一家公司的时候开始接触RxJava，当时才1.0版本，RxJava的越来越火了，现在最新的版本已经到了2.0版本了，对于1.0版本来说网上已经有很多的教程，而对于2.0版本还没有多少的教程介绍。写下这些笔记，以便以后回来温习或者可以对初始使用RxJava的朋友们有带来一些的入门。
附上RxJava1.0教程的干货：
给 Android 开发者的 RxJava 详解（抛物线大牛)
彻底了解RxJava
1、RxJava基础知识
RxJava是一个在JVM上使用可观测的序列来组成异步的，基于事件的程序库。简单的来说就是一个实现可以替换AsyncTask/Thread等异步操作的库。
2、RxJava2的核心概念
最核心的东西有一下几个：
Observable（被观察者）以及2.0出来的可缓存序列的被观察者Flowable支持非阻塞式的背压
Observer（观察者）或者叫Subscriber（订阅者)
Subscribe (订阅)、事件
两者之间通过一种扩展的观察者模式来实现关联，也叫做发布-订阅模式。RxJava把多个事件看做是一个队列，由Observable或者Flowable发出一系列的事件，并对每一个事件进行单独的处理，事件执行结束后交给Observer、Subscriber或者Consumer的回调进行处理。
对于RxJava2中Flowable和Subscriber和上面的1.0版本被观察者、观察者的使用和订阅关系类似的，只不过它支持背压，有兴趣可以看一下RxJava 2.0中backpressure(背压)概念的理解。
要在Android中使用RxJava2, 先添加Gradle配置:
compile 'io.reactivex.rxjava2:rxjava:2.0.0'
compile 'org.reactivestreams:reactive-streams:1.0.0'
和1.x中不一样，2.0有一个依赖包
4、RxJava2响应式编程结构
响应式编程的组成为Observable/Operator/Subscriber，RxJava在响应式编程中的基本流程如下：
这个流程，可以简单的理解为：Observable -> Operator1 -> Operator2 -> Operator3 -> Subscriber
1.Observable发出一系列事件，他是事件的产生者；
2.Subscriber负责处理事件，他是事件的消费者；
3.Operator是对Observable发出的事件进行修改和变换；
4. 若事件从产生到消费不需要其他处理，则可以省略掉中间的Operator，从而流程变为Obsevable -> Subscriber；
5. Subscriber通常在主线程执行，所以原则上不要去处理太多的事务，而这些复杂的处理则交给Operator；
5、小试RxJava2
1.创建一个Observable和一个Observer
public void createObservable() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
System.out.println("发射 A");
e.onNext("A");
System.out.println("发射 B");
e.onNext("B");
System.out.println("发射 C");
e.onNext("C");
e.onComplete();
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("onSubscribe");
}
@Override
public void onNext(String value) {
System.out.println("onNext:" + value);
}
@Override
public void onError(Throwable e) {
System.out.println("onError");
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
运行结果是：
onSubscribe
发射 A
onNext:A
发射 B
onNext:B
发射 C
onNext:C
onComplete
这里通过ObservableEmitter发射器发出三种类型的事件，onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、complete事件和error事件。
发射事件，需要满足一定的规则：
- 1.被观察者的发射器可以发送无限个onNext,而观察者也可以不断顺序接收多个事件；
- 2.当被观察者的发射器发送了一个onComplete后，发射器还可以继续发射onNext事件，而观察者接收到onComplete事件后将不再继续接收事件；例如上面程序发射顺序改成：
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
System.out.println("发射 A");
e.onNext("A");
System.out.println("发射 onComplete事件");
e.onComplete();
System.out.println("发射 B");
e.onNext("B");
System.out.println("发射 C");
e.onNext("C");
}
运行的结果是：
onSubscribe
发射 A
onNext:A
发射 onComplete事件
onComplete
发射 B
发射 C
- 3.同样，当被观察者的发射器发送了一个onError后，发射器还可以继续发射onNext事件，而观察者接收到onError事件后将不再继续接收事件；
- 4.对于onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError；
- 5.对于中途调用Disposable里面的disponse()方法，虽然不会影响发射器继续发射事件，但是会截断观察者不再接收剩余的事件；示例：
public void createObservableDispose() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
System.out.println("发射 A");
e.onNext("A");
System.out.println("发射 B");
e.onNext("B");
System.out.println("发射 C");
e.onNext("C");
System.out.println("发射 D");
e.onNext("D");
System.out.println("发射 onComplete事件");
e.onComplete();
}
}).subscribe(new Observer<String>() {
private Disposable disposable;
private int i;
@Override
public void onSubscribe(Disposable d) {
disposable = d;
System.out.println("onSubscribe");
}
@Override
public void onNext(String value) {
i++;
System.out.println(value);
if (i == 2) {
System.out.println("disponsable---");
disposable.dispose();
System.out.println("isDisponsable:" + disposable.isDisposed());
}
}
@Override
public void onError(Throwable e) {
System.out.println("onError");
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
运行结果：
onSubscribe
发射 A
A
发射 B
B
disponsable---
isDisponsable:true
发射 C
发射 D
发射 onComplete事件
以下补上泽毛大牛提到遗漏的两点：
- 6.上游调用了 onError 之后可以调用 onComplete，虽然下游收不到，但是不会报错；而如果上游调用了 onComplete 之后再调用 onError 就会报错了。测试示例:
public void createObservableErroComplete() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
System.out.println("发射 A");
e.onNext("A");
System.out.println("发射 B");
e.onNext("B");
System.out.println("发射 C");
e.onNext("C");
System.out.println("发射 onError事件");
e.onError(new NullPointerException());
System.out.println("发射 D");
e.onNext("D");
System.out.println("发射 onComplete事件");
e.onComplete();
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("onSubscribe");
}
@Override
public void onNext(String value) {
System.out.println(value);
}
@Override
public void onError(Throwable e) {
System.out.println("onError");
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
先发射onErro再发射onComplete不会报错，结果如下:
onSubscribe
发射 A
A
发射 B
B
发射 C
C
发射 onError事件
onError
发射 D
发射 onComplete事件
接着交换成先onComplete再onErro会报错:
public static void createObservableCompleteErro() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
System.out.println("发射 A");
e.onNext("A");
System.out.println("发射 B");
e.onNext("B");
System.out.println("发射 C");
e.onNext("C");
System.out.println("发射 onComplete事件");
e.onComplete();
System.out.println("发射 D");
e.onNext("D");
System.out.println("发射 onError事件");
e.onError(new NullPointerException());
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("onSubscribe");
}
@Override
public void onNext(String value) {
System.out.println(value);
}
@Override
public void onError(Throwable e) {
System.out.println("onError");
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
报错结果如下:
onSubscribe
java.lang.NullPointerException
发射 A
A
发射 B
B
发射 C
C
发射 onComplete事件
onComplete
发射 D
at com.jltx.rx.RxJava2Learn1$18.subscribe(RxJava2Learn1.java:377)
发射 onError事件
- 7.中途调用 Disposable 里面的 disponse() 方法之后，上游是可以继续发送 onNext 和 onComplete，但是不可以发送 onError，不然会报错的。示例:
public void createObservableDisposeErro() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
System.out.println("发射 A");
e.onNext("A");
System.out.println("发射 B");
e.onNext("B");
System.out.println("发射 C");
e.onNext("C");
System.out.println("发射 D");
e.onNext("D");
System.out.println("发射 onError事件");
e.onError(new NullPointerException());
}
}).subscribe(new Observer<String>() {
private Disposable disposable;
private int i;
@Override
public void onSubscribe(Disposable d) {
disposable = d;
System.out.println("onSubscribe");
}
@Override
public void onNext(String value) {
i++;
System.out.println(value);
if (i == 2) {
System.out.println("disponsable---");
disposable.dispose();
System.out.println("isDisponsable:" + disposable.isDisposed());
}
}
@Override
public void onError(Throwable e) {
System.out.println("onError");
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});
}
报错结果如下:
java.lang.NullPointerException
onSubscribe
at com.jltx.rx.RxJava2Learn1$20.subscribe(RxJava2Learn1.java:418)
发射 A
at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
A
at io.reactivex.Observable.subscribe(Observable.java:10151)
发射 B
at com.jltx.rx.RxJava2Learn1.createObservableDisposeErro(RxJava2Learn1.java:420)
B
at com.jltx.rx.RxJava2Learn1.main(RxJava2Learn1.java:35)
2.创建一个Flowable被观察者和一个Subscriber订阅者
public void createFlowable(){
// create a flowable
Flowable<String> flowable = Flowable.create(new FlowableOnSubscribe<String>() {
@Override
public void subscribe(FlowableEmitter<String> e) throws Exception {
e.onNext("Hello Rx2");
e.onComplete();
}
}, BackpressureStrategy.BUFFER);
// create subscriber
Subscriber subscriber = new Subscriber<String>(){
@Override
public void onSubscribe(Subscription s) {
System.out.println("onSubscribe");
//调用request去请求资源，参数就是要请求的数量，一般如果不限制请求数量，可以写成Long.MAX_VALUE。
// 如果你不调用request，Subscriber的onNext和onComplete方法将不会被调用
s.request(Long.MAX_VALUE);
}
@Override
public void onNext(String s) {
System.out.println("onNext:"+s);
}
@Override
public void onError(Throwable t) {
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
};
flowable.subscribe(subscriber);
}
这里创建Flowable的时候传入背压策略来应对当一个快速发送消息的被观察者遇到一个处理消息缓慢的观察者，这里默认传入BackpressureStrategy.BUFFER给observable发送出来的数据持有一个缓存，当request方法被调用的时候，给下层流发送一个item。
这里需要注意的是在onSubscribe中，我们需要调用request去请求资源，参入的参数就是需要请求的资源的限制数量，写成Long.MAX_VALUE相当于不限制数量，如果不调用request；Subscriver的onNext和onComplete方法不会被调用。
