前言
Rxjava由于其基于事件流的链式调用、逻辑简洁 & 使用简单的特点，深受各大 Android开发者的欢迎。
如果还不了解RxJava，请看文章：Android：这是一篇 清晰 & 易懂的Rxjava 入门教程
- 今天，我将为大家带来 源码分析：
Rxjava的订阅流程，其为Rxjava使用的基本 & 核心，希望大家会喜欢。
Carson带你学RxJava系列文章，包括 原理、操作符、应用场景、背压等等，请关注看文章：Android：这是一份全面 & 详细的RxJava学习指南
目录
示意图
1. RxJava简介
此处简单介绍RxJava
示意图
若还不了解RxJava，请看文章：Android：这是一篇 清晰 & 易懂的Rxjava 入门教程
2. 订阅流程 的使用
2.1 使用步骤
RxJava的订阅流程 使用方式 = 基于事件流的链式调用，具体步骤如下：
步骤1：创建被观察者(Observable)& 定义需发送的事件
步骤2：创建观察者(Observer) & 定义响应事件的行为
步骤3：通过订阅（subscribe）连接观察者和被观察者
2.2 实例讲解
// RxJava的链式操作
Observable.create(new ObservableOnSubscribe<Integer>() {
// 1. 创建被观察者(Observable) & 定义需发送的事件
@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. 创建观察者(Observer) & 定义响应事件的行为
// 3. 通过订阅（subscribe）连接观察者和被观察者
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）
@Override
public void onNext(Integer value) {
Log.d(TAG, "对Next事件"+ value +"作出响应" );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
}
}
- 运行结果
示意图
3. 源码分析
下面，我将根据 使用步骤 进行RxJava的源码分析：
步骤1：创建被观察者(Observable)& 定义需发送的事件
步骤2：创建观察者(Observer) & 定义响应事件的行为
步骤3：通过订阅（subscribe）连接观察者和被观察者
步骤1：创建被观察者(Observable)& 定义需发送的事件
- 源码分析如下
/**
* 使用步骤1：创建被观察者(Observable)& 定义需发送的事件
**/
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();
}
})
/**
* 源码分析：Observable.create(new ObservableOnSubscribe<Integer>(){...}）
**/
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
...
// 仅贴出关键源码
return new ObservableCreate<T>(source);
// 创建ObservableCreate类对象 ->>分析1
// 注：传入source对象（即 我们手动创建的ObservableOnSubscribe对象）
}
/**
* 分析1：new ObservableCreate<T>(source)
**/
public final class ObservableCreate<T> extends Observable<T> {
// ObservableCreate类 = Observable的子类
...
// 仅贴出关键源码
final ObservableOnSubscribe<T> source;
// 构造函数
// 传入了传入source对象 = 手动创建的ObservableOnSubscribe对象
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
/**
* 重点关注：复写了subscribeActual（）
* 作用：订阅时，通过接口回调 调用被观察者（Observerable） 与 观察者（Observer）的方法
**/
@Override
protected void subscribeActual(Observer<? super T> observer) {
// 1. 创建1个CreateEmitter对象（封装成1个Disposable对象）
// 作用：发射事件
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
// 2. 调用观察者（Observer）的onSubscribe（）
// onSubscribe（）的实现 = 使用步骤2（创建观察者（Observer））时复写的onSubscribe(）
observer.onSubscribe(parent);
try {
// 3. 调用source对象的subscribe（）
// source对象 = 使用步骤1（创建被观察者（Observable））中创建的ObservableOnSubscribe对象
// subscribe（）的实现 = 使用步骤1（创建被观察者（Observable））中复写的subscribe（）->>分析2
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
/**
* 分析2：emitter.onNext("1");
* 此处仅讲解subscribe（）实现中的onNext（）
* onError（）、onComplete()类似，此处不作过多描述
**/
static final class CreateEmitter<T> extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
...
// 仅贴出关键代码
// onNext（）源码分析
@Override
public void onNext(T t) {
// 注：发送的事件不可为空
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
// 若无断开连接（调用Disposable.dispose()），则调用观察者（Observer）的同名方法 = onNext（）
// 观察者的onNext（）的内容 = 使用步骤2中复写内容
if (!isDisposed()) {
observer.onNext(t);
}
}
// onError（）、onComplete()类似，此处不作过多描述
// 特别说明：调用该2方法，最终都会自动调用dispose()，即断开观察者 & 被观察者的连接
@Override
public void onError(Throwable t) {
if (t == null) {
t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
}
if (!isDisposed()) {
try {
observer.onError(t);
} finally {
dispose();
}
} else {
RxJavaPlugins.onError(t);
}
}
@Override
public void onComplete() {
if (!isDisposed()) {
try {
observer.onComplete();
} finally {
dispose();
}
}
}
- 步骤1总结
示意图
步骤2：创建观察者 & 定义响应事件的行为
- 源码分析
/**
* 使用步骤2：创建观察者 & 定义响应事件的行为（方法内的创建对象代码）
**/
subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）
@Override
public void onNext(Integer value) {
Log.d(TAG, "对Next事件"+ value +"作出响应" );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
/**
* 源码分析：Observer类
**/
public interface Observer<T> {
// 注：Observer本质 = 1个接口
// 接口内含4个方法，分别用于 响应 对应于被观察者发送的不同事件
void onSubscribe(@NonNull Disposable d); // 内部参数：Disposable 对象，可结束事件
void onNext(@NonNull T t);
void onError(@NonNull Throwable e);
void onComplete();
}
/**
* 特别说明：Subscriber类
* 定义：RxJava 内置的一个实现了 Observer 的抽象类
* 作用：扩展Observer 接口 = 新增了2个方法 =
* 1. onStart()：在还未响应事件前调用，用于初始化工作
* 2. unsubscribe()：用于取消订阅。在该方法被调用后，观察者将不再接收 & 响应事件
* 注：调用该方法前，先使用 isUnsubscribed() 判断状态，确定被观察者Observable是否还持有观察者Subscriber的引用；若引用不能及时释放，就会出现内存泄露
* 使用方式：与Observer使用几乎相同（实质上，Observer总是会先被转换成Subscriber再使用）
**/
Subscriber<String> subscriber = new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "开始采用subscribe连接");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "对Next事件作出响应" + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
};
步骤3：通过订阅（subscribe）连接观察者和被观察者
- 源码分析
/**
* 使用步骤3：通过订阅（subscribe）连接观察者和被观察者 = subscribe（）
**/
subscribe(new Observer<Integer>() {
// 2. 通过通过订阅（subscribe）连接观察者和被观察者
// 3. 创建观察者 & 定义响应事件的行为
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）
@Override
public void onNext(Integer value) {
Log.d(TAG, "对Next事件"+ value +"作出响应" );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
/**
* 源码分析：Observable.subscribe（observer）
* 说明：该方法属于 Observable 类的方法（注：传入1个 Observer 对象）
**/
@Override
public final void subscribe(Observer<? super T> observer) {
...
// 仅贴出关键源码
subscribeActual(observer);
// 继续往下看：分析1
}
/**
* Observable.subscribeActual(observer)
* 说明：属于抽象方法，由子类实现；此处的子类 = 步骤1创建被观察者（Observable）时创建的ObservableCreate类
* 即 在订阅时，实际上是调用了步骤1创建被观察者（Observable）时创建的ObservableCreate类里的subscribeActual（）
* 此时，你应该回头看上面的步骤1里的subscribeActual()，应该能理解RxJava的整个订阅流程了。
**/
protected abstract void subscribeActual(Observer<? super T> observer);
-
总结
示意图
4. 源码总结
- 在步骤1（创建被观察者（
Observable））、步骤2（创建观察者（Observer））时，仅仅只是定义了发送的事件 & 响应事件的行为； - 只有在步骤3（订阅时），才开始发送事件 & 响应事件，真正连接了被观察者 & 观察者
- 具体源码总结如下
示意图
5. 特别注意：涉及多个被观察者（Observable）的发送事件顺序
- 具体描述
示意图
- 实例讲解
/**
* 存在涉及多个被观察者（Observable）的情况
**/
// 创建第1个被观察者（Observable1）
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
})
// 使用flatMap操作符（内部会创建第2个被观察者（Observable2））
.flatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Integer integer) throws Exception {
final List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("我是事件" + integer + "拆分后的子事件" + i);
// 通过flatMap中将被观察者生产的事件序列先进行拆分，再将每个事件转换为一个新的发送三个String事件
// 最终合并，再发送给被观察者
}
return Observable.fromIterable(list);
}
})
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）
@Override
public void onNext(String value) {
Log.d(TAG, "响应事件："+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
// 过程讲解
// 调用顺序：先回调Observable2的subscribe(Observer) 、subscribeActual(Observer)、再回调Observable1的subscribe(Observer) 、subscribeActual(Observer)
// Observable的发送顺序 = 先发送Observable1、再发送Observable2
-
测试结果
示意图
6. 总结
本文主要对
RxJava2中 的订阅流程进行了源码分析Carson带你学RxJava系列文章：
入门
Carson带你学Android：这是一篇清晰易懂的Rxjava入门教程
Carson带你学Android：面向初学者的RxJava使用指南
Carson带你学Android：RxJava2.0到底更新了什么？
原理
Carson带你学Android：图文解析RxJava原理
Carson带你学Android：手把手带你源码分析RxJava
使用教程：操作符
Carson带你学Android：RxJava操作符教程
Carson带你学Android：RxJava创建操作符
Carson带你学Android：RxJava功能性操作符
Carson带你学Android：RxJava过滤操作符
Carson带你学Android：RxJava组合/合并操作符
Carson带你学Android：RxJava变换操作符
Carson带你学Android：RxJava条件/布尔操作符
实战
Carson带你学Android：什么时候应该使用Rxjava？(开发场景汇总)
Carson带你学Android：RxJava线程控制(含实例讲解)
Carson带你学Android：图文详解RxJava背压策略
Carson带你学Android：RxJava、Retrofit联合使用汇总(含实例教程)
Carson带你学Android：优雅实现网络请求嵌套回调
Carson带你学Android：网络请求轮询(有条件)
Carson带你学Android：网络请求轮询(无条件)
Carson带你学Android：网络请求出错重连(结合Retrofit)
Carson带你学Android：合并数据源
Carson带你学Android：联想搜索优化
Carson带你学Android：功能防抖
Carson带你学Android：从磁盘/内存缓存中获取缓存数据
Carson带你学Android：联合判断
欢迎关注Carson_Ho的简书
不定期分享关于安卓开发的干货，追求短、平、快，但却不缺深度。
