一、 RxJava简介
（一） 什么是RxJava
“a library for composing asynchronous and event-based programs using observable sequences for the Java VM”
一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库。它是一个用于处理异步的框架，
（二） 为什么使用RxJava
异步操作很关键的一点是程序的简洁性，因为在调度过程比较复杂的情况下，异步代码经常会既难写也难被读懂。 Android 创造的 AsyncTask 和Handler ，其实都是为了让异步代码更加简洁。RxJava 的优势也是简洁，但它的简洁的与众不同之处在于，随着程序逻辑变得越来越复杂，它依然能够保持简洁。
（三）RxJava的基本概念
基本概念
即RxJava原理可总结为：被观察者 （Observable） 通过 订阅（Subscribe） 按顺序发送事件 给观察者 （Observer）， 观察者（Observer） 按顺序接收事件 & 作出对应的响应动作。
流程图
二、 RxJava的简单使用
（一）配置Gradle文件添加依赖
implementation 'io.reactivex.rxjava2:rxjava:2.1.16'
implementation 'io.reactivex.rxjava2:rxandroid:2.0.2'
（二）基本实现
1. 创建一个被观察者
// 1. 创建被观察者 Observable 对象
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
// create() 是 RxJava 最基本的创造事件序列的方法
// 此处传入了一个 OnSubscribe 对象参数
// 当 Observable 被订阅时，OnSubscribe 的 call() 方法会自动被调用，即事件序列就会依照设定依次被触发
// 即观察者会依次调用对应事件的复写方法从而响应事件
// 从而实现被观察者调用了观察者的回调方法 & 由被观察者向观察者的事件传递，即观察者模式
// 2. 在复写的subscribe（）里定义需要发送的事件
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// 通过 ObservableEmitter类对象产生事件并通知观察者
// ObservableEmitter类介绍
// a. 定义：事件发射器
// b. 作用：定义需要发送的事件 & 向观察者发送事件
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
});
2. 创建一个观察者
<--方式一：实现Observer 接口 -->
// 1. 创建Observer 对象
Observer<Integer> observer = new Observer<Integer>() {
// 观察者接收事件前，默认最先调用复写 onSubscribe（）
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "采用subscribe连接");
}
// 当被观察者生产Next事件 & 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onNext(Integer value) {
Log.d(TAG, "对Next事件作出响应" + value);
}
// 当被观察者生产Error事件& 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
// 当被观察者生产Complete事件& 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
};
<--方式二：采用Subscriber 抽象类 -->
// 说明：Subscriber类 = RxJava 内置的一个实现了 Observer 的抽象类，对 Observer 接口进行了扩展
// 1. 创建观察者 （Observer ）对象
Subscriber<String> subscriber = new Subscriber<Integer>() {
// 2. 创建对象时通过对应复写对应事件方法 从而 响应对应事件
// 观察者接收事件前，默认最先调用复写 onSubscribe（）
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "开始采用subscribe连接");
}
// 当被观察者生产Next事件 & 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onNext(Integer value) {
Log.d(TAG, "对Next事件作出响应" + value);
}
// 当被观察者生产Error事件& 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
// 当被观察者生产Complete事件& 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
};
Subscriber 抽象类与Observer 接口的区别
相同点：二者基本使用方式完全一致（实质上，在RxJava的 subscribe 过程中，Observer总是会先被转换成Subscriber再使用）
不同点：Subscriber抽象类对 Observer 接口进行了扩展，新增了两个方法：
- onStart()：在还未响应事件前调用，用于做一些初始化工作
- unsubscribe()：用于取消订阅。在该方法被调用后，观察者将不再接收 & 响应事件
调用该方法前，先使用 isUnsubscribed() 判断状态，确定被观察者Observable是否还持有观察者Subscriber的引用，如果引用不能及时释放，就会出现内存泄露
3. 被观察者和观察者订阅（建立连接）
observable.subscribe(observer);
（三） 链式连接
Observer大部分操作方法都会返回一个新的Observer,所以上述代码可以合并为一句代码，也就是RxJava所说的链式调用。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG,"开始采用subscribe连接");
}
@Override
public void onNext(Integer integer) {
Log.d(TAG,"对事件"+integer+"响应");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
运行结果
