1 前言
Rxjava由于其基于事件流的链式调用、逻辑简洁 & 使用简单的特点，深受各大 Android开发者的欢迎。
RxJava github地址
944365-4c1c1eb44ffe01e5.png
1 定义
- RxJava 在 GitHub 的介绍：
RxJava：a library for composing asynchronous and event-based programs using observable sequences for the Java VM
// 翻译：RxJava 是一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库
- 总结：RxJava 是一个 基于事件流、实现异步操作的库
2 作用
- 实现异步操作
- 类似于 Android中的 AsyncTask 、Handler作用
3 特点
由于 RxJava的使用方式是：基于事件流的链式调用，所以使得 RxJava：
- 逻辑简洁
- 实现优雅
- 使用简单
更重要的是，随着程序逻辑的复杂性提高，它依然能够保持简洁 & 优雅
4 原理
4.1 生活例子引入
- 顾客到饭店吃饭
顾客到饭店吃饭.png
944365-07f12da4616b2b68.png
4.2 Rxjava原理介绍
- Rxjava原理 基于.一种扩展的观察者模式
- Rxjava的扩展观察者模式中有4个角色
| 角色 | 作用 | 类比 |
|---|---|---|
| 被观察者（Observable） | 产生事件 | 顾客 |
| 观察者（Observer） | 接收事件，并给出响应动作 | 厨房 |
| 订阅（Subscribe） | 连接 被观察者 & 观察者 | 服务员 |
| 事件（Event） | 被观察者 & 观察者 沟通的载体 | 菜式 |
具体原理
请结合上述 顾客到饭店吃饭 的生活例子理解：
944365-5b6e7c8a3bb55f39.png
944365-fc3b7eb5a0ad28d0.png
- 即RxJava原理可总结为：被观察者 （Observable） 通过 订阅（Subscribe）按顺序发送事件 给观察者 （Observer）， 观察者（Observer） 按顺序接收事件 & 作出对应的响应动作。具体如下图：
944365-98ec92df0a4d7e0b.png
至此，RxJava原理讲解完毕。
5 基本使用
- 1.分步骤实现：该方法主要为了深入说明Rxjava的原理 & 使用，主要用于演示说明
- 2.基于事件流的链式调用：主要用于实际使用
5.1 方式1：分步骤实现
5.1.1 使用步骤
944365-779b0832b164e116.png
5.1.2 步骤详解
//步骤1：创建被观察者 （Observable ）& 生产事件
Observable<Integer> observable = createObservable1();
//步骤2：创建观察者 （Observer ）并 定义响应事件的行为
Observer<Integer> observer = createObserver1();
//步骤3：通过订阅（Subscribe）连接观察者和被观察者
observable.subscribe(observer);
// 或者 observable.subscribe(subscriber)；
步骤1：创建被观察者 （Observable ）& 生产事件
- 即 顾客入饭店 - 坐下餐桌 - 点菜
- 具体实现
create()--just(T...)--fromXX()
create()
private Observable<Integer> createObservable1() {
// 1. 创建被观察者 Observable 对象
// create() 是 RxJava 最基本的创造事件序列的方法
// 此处传入了一个 OnSubscribe 对象参数
// 当 Observable 被订阅时，OnSubscribe 的 call() 方法会自动被调用，即事件序列就会依照设定依次被触发
// 即观察者会依次调用对应事件的复写方法从而响应事件
// 从而实现被观察者调用了观察者的回调方法 & 由被观察者向观察者的事件传递，即观察者模式
Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
// 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();
}
});
return integerObservable;
}
just(T...)
//扩展：RxJava 提供了其他方法用于 创建被观察者对象Observable
// 方法1：just(T...)：直接将传入的参数依次发送出来
private Observable createObservable2() {
Observable observable = Observable.just("A", "B", "C");
return observable;
from(T[])
//扩展：RxJava 提供了其他方法用于 创建被观察者对象Observable
// 方法2：from(T[]) 将传入的数组 / Iterable 拆分成具体对象后，依次发送出来
private Observable createObservable3() {
String[] words = {"A", "B", "C"};
Observable observable = Observable.fromArray(words);
return observable;
}
步骤2：创建观察者 （Observer ）并 定义响应事件的行为
- 即 开厨房 - 确定对应菜式
- 发生的事件类型包括：Next事件、Complete事件 & Error事件。具体如下：
944365-8cb0da34f94b0c73.png
具体实现
- 方式1：采用Observer 接口
- 方式2：采用Subscriber 抽象类
2种方法的区别- 相同点：二者基本使用方式完全一致（实质上，在RxJava的 subscribe 过程中，Observer总是会先被转换成Subscriber再使用）
- 不同点：Subscriber抽象类对 Observer 接口进行了扩展，新增了两个方法：
- onStart()：在还未响应事件前调用，用于做一些初始化工作
- unsubscribe()：用于取消订阅。在该方法被调用后，观察者将不再接收 & 响应事件
- 调用该方法前，先使用 isUnsubscribed() 判断状态，确定被观察者Observable是否还持有观察者Subscriber的引用，如果引用不能及时释放，就会出现内存泄露
采用Observer 接口
private Observer<Integer> createObserver1() {
//方式1：采用Observer 接口
// 1. 创建观察者 （Observer ）对象
Observer<Integer> observer = new Observer<Integer>() {
// 2. 创建对象时通过对应复写对应事件方法 从而 响应对应事件
// 观察者接收事件前，默认最先调用复写 onSubscribe（）
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 当被观察者生产Next事件 & 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onNext(Integer integer) {
Log.d(TAG, "对Next事件作出响应" + integer);
}
// 当被观察者生产Error事件& 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
// 当被观察者生产Complete事件& 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
};
return observer;
}
方式2：采用Subscriber 抽象类
private Subscriber<Integer> createObserver2() {
//方式2：采用Subscriber 抽象类
// 说明：Subscriber类 = RxJava 内置的一个实现了 Observer 的抽象类，
// 对 Observer 接口进行了扩展\
// 1. 创建观察者 （Observer ）对象
Subscriber<Integer> subscriber = new Subscriber<Integer>() {
// 2. 创建对象时通过对应复写对应事件方法 从而 响应对应事件
// 观察者接收事件前，默认最先调用复写 onSubscribe（）
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "开始采用subscribe连接");
}
// 当被观察者生产Next事件 & 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onNext(Integer integer) {
Log.d(TAG, "对Next事件作出响应" + integer);
}
// 当被观察者生产Error事件& 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onError(Throwable t) {
Log.d(TAG, "对Error事件作出响应");
}
// 当被观察者生产Complete事件& 观察者接收到时，会调用该复写方法 进行响应
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
};
return subscriber;
}
步骤3：通过订阅（Subscribe）连接观察者和被观察者
- 即 顾客找到服务员 - 点菜 - 服务员下单到>厨房 - 厨房烹调
- 具体实现
observable.subscribe(observer);
// 或者 observable.subscribe(subscriber)；
5.2 方式2：优雅的实现方法 - 基于事件流的链式调用
- 上述的实现方式是为了说明Rxjava的原理 & 使用
- 在实际应用中，会将上述步骤&代码连在一起，从而更加简洁、更加优雅，即所谓的 RxJava基于事件流的链式调用.
private void chainCall1() {
// RxJava的链式操作
// 1. 创建被观察者 & 生产事件
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>() {
// 2. 通过通过订阅（subscribe）连接观察者和被观察者
// 3. 创建观察者 & 定义响应事件的行为
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "对Next事件" + integer + "作出响应");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
//注：整体方法调用顺序：观察者.onSubscribe（）> 被观察者.subscribe（）> 观察者.onNext（）>观察者.onComplete()
}
RxJava 2.x 提供了多个函数式接口 ，用于实现简便式的观察者模式。
944365-abda1c2bef8681f3.png
private void chainCall2() {
Observable.just("hello").subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d(TAG, s);
}
});
}
6 额外说明
6.1 观察者 Observer的subscribe()具备多个重载的方法
public final Disposable subscribe() {}
// 表示观察者不对被观察者发送的事件作出任何响应（但被观察者还是可以继续发送事件）
public final Disposable subscribe(Consumer<? super T> onNext) {}
// 表示观察者只对被观察者发送的Next事件作出响应
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {}
// 表示观察者只对被观察者发送的Next事件 & Error事件作出响应
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
// 表示观察者只对被观察者发送的Next事件、Error事件 & Complete事件作出响应
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
// 表示观察者只对被观察者发送的Next事件、Error事件 、Complete事件 & onSubscribe事件作出响应
public final void subscribe(Observer<? super T> observer) {}
// 表示观察者对被观察者发送的任何事件都作出响应
6.2 可采用 Disposable.dispose() 切断观察者 与 被观察者 之间的连接
即观察者 无法继续 接收 被观察者的事件，但被观察者还是可以继续发送事件
// 主要在观察者 Observer中 实现
Observer<Integer> observer = new Observer<Integer>() {
// 1. 定义Disposable类变量
private Disposable mDisposable;
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
// 2. 对Disposable类变量赋值
mDisposable = d;
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "对Next事件"+ value +"作出响应" );
if (value == 2) {
// 设置在接收到第二个事件后切断观察者和被观察者的连接
mDisposable.dispose();
Log.d(TAG, "已经切断了连接：" + mDisposable.isDisposed());
}
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
};
