Android RxJava：一步步带你源码分析 RxJava
16lz
2021-01-23
前言
Rxjava由于其基于事件流的链式调用、逻辑简洁 & 使用简单的特点，深受各大Android开发者的欢迎。
如果还不了解RxJava，请看文章：Android：这是一篇 清晰 & 易懂的Rxjava 入门教程
- 今天，我将为大家带来 源码分析：
Rxjava的订阅流程，其为Rxjava使用的基本 & 核心，希望大家会喜欢。
- 本系列文章主要基于
Rxjava 2.0- 接下来的时间，我将持续推出
Android中Rxjava 2.0的一系列文章，包括原理、操作符、应用场景、背压等等 ，有兴趣可以继续关注Carson_Ho的安卓开发笔记！！
目录
1. RxJava简介
此处简单介绍RxJava
若还不了解RxJava，请看文章：Android：这是一篇 清晰 & 易懂的Rxjava 入门教程
2. 订阅流程 的使用
2.1 使用步骤
RxJava的订阅流程 使用方式 = 基于事件流的链式调用，具体步骤如下：
步骤1：创建被观察者(Observable)& 定义需发送的事件
步骤2：创建观察者(Observer) & 定义响应事件的行为
步骤3：通过订阅（subscribe）连接观察者和被观察者
2.2 实例讲解
// RxJava的链式操作 Observable.create(new ObservableOnSubscribe() { // 1. 创建被观察者(Observable) & 定义需发送的事件 @Override public void subscribe(ObservableEmitter emitter) throws Exception { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onComplete(); } }).subscribe(new Observer() { // 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() { @Override public void subscribe(ObservableEmitter emitter) throws Exception { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onComplete(); } })/** * 源码分析：Observable.create(new ObservableOnSubscribe(){...}） **/public static Observable create(ObservableOnSubscribe source) { ... // 仅贴出关键源码 return new ObservableCreate(source); // 创建ObservableCreate类对象 ->>分析1 // 注：传入source对象（即 我们手动创建的ObservableOnSubscribe对象） } /** * 分析1：new ObservableCreate(source) **/public final class ObservableCreate extends Observable { // ObservableCreate类 = Observable的子类 ... // 仅贴出关键源码 final ObservableOnSubscribe source; // 构造函数 // 传入了传入source对象 = 手动创建的ObservableOnSubscribe对象 public ObservableCreate(ObservableOnSubscribe source) { this.source = source; } /** * 重点关注：复写了subscribeActual（） * 作用：订阅时，通过接口回调 调用被观察者（Observerable） 与 观察者（Observer）的方法 **/ @Override protected void subscribeActual(Observer<? super T> observer) { // 1. 创建1个CreateEmitter对象（封装成1个Disposable对象） // 作用：发射事件 CreateEmitter parent = new CreateEmitter(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 extends AtomicReference implements ObservableEmitter, 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() { @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 { // 注：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 subscriber = new Subscriber() { @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() { // 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() { @Override public void subscribe(ObservableEmitter emitter) throws Exception { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); } }) // 使用flatMap操作符（内部会创建第2个被观察者（Observable2）） .flatMap(new Function>() { @Override public ObservableSource apply(Integer integer) throws Exception { final List list = new ArrayList<>(); for (int i = 0; i < 3; i++) { list.add("我是事件" + integer + "拆分后的子事件" + i); // 通过flatMap中将被观察者生产的事件序列先进行拆分，再将每个事件转换为一个新的发送三个String事件 // 最终合并，再发送给被观察者 } return Observable.fromIterable(list); } }) .subscribe(new Observer() { @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中 的订阅流程进行了源码分析 -
接下来的时间，我将持续推出
Android中Rxjava 2.0的一系列文章，包括原理、操作符、应用场景、背压等等 -
感兴趣的同学可以继续关注carson_ho的微信公众号
请帮顶 / 评论点赞！因为你的鼓励是我写作的最大动力！
更多相关文章
- 【Android基础】讲讲Android的事件分发机制
- android JNI 开发步骤
- Android之View篇2————View的事件分发
- 事件机制（2）--Android onTouchEvent, onClick及onLongClick的调用
- android的touch事件详解