小白读源码 | RxJava2 入门篇(一)
转载请注明出处： juejin.im/user/590207…
题记： RxJava2 想必很多人都用过，扩展的观察者模式，简洁的链式调用，通过简单的API调用就可以满足我们的各种需求，让人不禁感叹这玩意儿真爽。当然在我们用着很爽的时候，不禁也会对它产生一些好奇，这玩意儿到底长是个啥模样，嗯，想看看，那就看看吧。花了些时间看了看它的部分源码，作此记录。
引子
既然我是只小白，还挑什么呢，捡最容易的上手噻，Flowable (带背压模式的被观察者），我还没有看， 这里仅记录普通的 Observable 源码阅读过程 。下面代码即为众所周知的入门用法，本篇文章就围绕它来阐述。这是我使用的版本：
compile 'io.reactivex.rxjava2:rxjava:2.1.1' compile 'io.reactivex.rxjava2:rxandroid:2.1.0' 复制代码
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "Observable emit 1 ");
emitter.onNext(1);
Log.d(TAG, "Observable emit 2");
emitter.onNext(2);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.d(TAG, "onSubscribe: isDisposable " + d.isDisposed());
}
@Override
public void onNext(@NonNull Integer integer) {
Log.d(TAG, "onNext: " + integer);
}
@Override
public void onError(@NonNull Throwable e) {
Log.d(TAG, "onError: " + e.getMessage());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: ");
}
});
复制代码
一 、关键概念
Observable
（被观察者）：这是一个抽象类，里面方法众多，就不列举了，读的时候遇到哪个看哪个。 Observer
（观察者）：这是个接口，里面有 4 个方法，是必须都要知道的。
public interface Observer<T> {
void onSubscribe(@NonNull Disposable d);
void onNext(@NonNull T t);
void onError(@NonNull Throwable e);
void onComplete();
}
复制代码
ObservableOnSubscribe
（事件发射器的载体）：
public interface ObservableOnSubscribe<T> {
void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception;
}
复制代码
ObservableEmitter
（事件发射器）：这是个接口，继承了 Emitter 接口，用于发送事件。
public interface ObservableEmitter<T> extends Emitter<T> {
void setDisposable(@Nullable Disposable d);
void setCancellable(@Nullable Cancellable c);
boolean isDisposed();
@NonNull
ObservableEmitter<T> serialize();
boolean tryOnError(@NonNull Throwable t);
}
public interface Emitter<T> {
void onNext(@NonNull T value);
void onError(@NonNull Throwable error);
void onComplete();
}
复制代码
这些关键概念必须要记住，至少大体知道都是什么，里面都有些什么方法。
二、直奔核心
既然这是个链式调用，我们不妨从头到尾过一遍。 Observable.create(new ObservableOnSubscribe<Integer>(){...})
创建了一个Observable对象，那就进 create()
这个静态方法看一看
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
// 判空代码，不重要，不看也罢
ObjectHelper.requireNonNull(source, "source is null");
// 创建 Observable 对象的关键代码
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
复制代码
create()
方法里只有两行代码，我们重点来看看 RxJavaPlugins.onAssembly(new ObservableCreate<T>(source))
这行代码 。很明显分为两部分， onAssembly()
和 new ObservableCreate<T>(source)
，我们先看 onAssembly()
，点进去发现是
@SuppressWarnings({ "rawtypes", "unchecked" })
@NonNull
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
// 上面两行代码是和使用 map 操作符相关的，我们这里还没用 map 操作符呢，对我们没啥卵用，当它不存在，
// 那么方法参数里传进来一个 Observable 类型的 source, 现在被原原本本当做返回值返回回去
// 意思就很明显，Observable 对象是由 new ObservableCreate<T>(source) 生成的
return source;
}
复制代码
既然 onAssembly()
把参数原样作为返回值返回，那 Observable.create(new ObservableOnSubscribe<Integer>(){...})
创建的 Observable
对象就是 new ObservableCreate<T>(source)
了，那我们就来看 new ObservableCreate<T>(source)
，点进去看关键代码
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
// 方法名暴露了真相 "实际订阅，真实订阅"
@Override
protected void subscribeActual(Observer<? super T> observer) {
// 1.创建 CreateEmitter 对象，参数传的是 observer， 这个 observer 从哪冒出来的，待会儿就知道了
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
// 2.并且这个 observer 还订阅了 CreateEmitter 对象
observer.onSubscribe(parent);
try {
// 3.source 也订阅了 CreateEmitter 对象
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
...
}
复制代码
首先 ObservableCreate
继承 Observable
,那 ObservableCreate
就是个 Observable
了，对，就是这么简单， ObservableCreate
就是我们要找的 Observable
对象 。然后上面 subscribeActual()
方法里添加注释的那 3 行代码讲的很清楚，一个是 CreateEmitter
（发射器），一个是 observer
（观察者），一个是 source
（这个 source
就是 ObservableOnSubscribe
，下面我就以 ObservableOnSubscribe
指代 source
），看到这 3 行代码，我们就敢假设整个观察者模式的消息订阅与发布就是由这 3 行代码控制的，要验证假设，我们还需再往下读源码。既然 observer
和 ObservableOnSubscribe
都与 CreateEmitter
有关，我们就来看看 CreateEmitter
，
static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
private static final long serialVersionUID = -3434801548987643227L;
final Observer<? super T> observer;
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
@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;
}
// 每次一定会先判断连接有没有切断（就是有没有 dispose），没有切断才接收事件
// 这这个判断就保证了一旦切断肯定就收不到事件了
if (!isDisposed()) {
observer.onNext(t);
}
}
@Override
public void onError(Throwable t) {
// 如果连接已经切断，还调用 Observer 的 onError() 方法，那就抛异常了
if (!tryOnError(t)) {
RxJavaPlugins.onError(t);
}
}
@Override
public boolean tryOnError(Throwable t) {
if (t == null) {
t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
}
// 如果连接没有切断，就调用 Observer 的 onError() 方法
if (!isDisposed()) {
try {
observer.onError(t);
} finally {
// 如果连接没有切断，在调用 Observer 的 onError() 方法后，一定会调用 dispose() 切断连接
dispose();
}
return true;
}
return false;
}
@Override
public void onComplete() {
// // 如果连接没有切断，就调用 Observer 的 onComplete() 方法
if (!isDisposed()) {
try {
observer.onComplete();
} finally {
// 如果连接没有切断，在调用 Observer 的 onComplete() 方法后，一定会调用 dispose() 切断连接
dispose();
}
}
}
...
}
复制代码
可以看到，
CreateEmitter
继承 ObservableEmitter<T>
和 Disposable
，那它就既是个 ObservableEmitter
，又是个 Disposable
，那它什么时候是 ObservableEmitter
，又什么时候是 Disposable
呢，当然是
在 observer.onSubscribe(parent)
里它是 Disposable
，在 source.subscribe(parent)
里它是 ObservableEmitter
。
为什么这么说呢，我们再接着看你就明白了。这里我先讲讲 source.subscribe(parent)
，其实这句代码就是 ObservableOnSubscribe.subscribe(ObservableEmitter)
,再看清楚些就是
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
...
}
}).subscribe(new Observer<Integer>() {
...
});
复制代码
上面几行代码就能解释
“ 在 source.subscribe(parent)
里它是 ObservableEmitter
”
这句话 。
接下来看连接 Observable
和 Observer
的 subscribe()
方法，
@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
// 判空，不用看
ObjectHelper.requireNonNull(observer, "observer is null");
try {
// 1.这个其实没啥用，我们用的是最简单的用法，所以参数传的是什么，返回值将它原样返回
observer = RxJavaPlugins.onSubscribe(this, observer);
// 判空，不用看
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
// 2.关键，进入这个方法发现它是 Observale 类里的一个抽象方法，这个抽象方法在哪里实现呢，
// 就在创建 Observable 对象的 create() 方法里的 new ObservableCreate<T>(source) 里
// 不信可以翻看上面介绍 new ObservableCreate<T>(source) 的代码
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
...
}
}
复制代码
subscribe()
方法里的 subscribeActual(observer)
方法在 new ObservableCreate<T>(source)
里重写了，翻看 new ObservableCreate<T>(source)
的代码后，你会发现它里面的 subscribeActual
方法里的 observer
就是 subscribe()
方法里的 observer
。
// 方法名暴露了真相 "实际订阅，真实订阅"
@Override
protected void subscribeActual(Observer<? super T> observer) {
// 1.创建 CreateEmitter 对象，参数传的是 observer， 这个 observer 从哪冒出来的，待会儿就知道了
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
// 2.并且这个 observer 还订阅了 CreateEmitter 对象
observer.onSubscribe(parent);
try {
// 3.source 也订阅了 CreateEmitter 对象
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
复制代码
既然 subscribe()
方法里的 observer
在 subscribeActual
方法里执行了 observer.onSubscribe(parent)
，那我们就来看下 subscribe()
方法的参数 Observer
对象吧，
...subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.d(TAG, "onSubscribe: isDisposable " + d.isDisposed());
}
@Override
public void onNext(@NonNull Integer integer) {
Log.d(TAG, "onNext: " + integer);
}
@Override
public void onError(@NonNull Throwable e) {
Log.d(TAG, "onError: " + e.getMessage());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete: ");
}
}
...
复制代码
看没看到 public void onSubscribe(@NonNull Disposable d)
, 在这里就可以解释
“ 在 observer.onSubscribe(parent)
里它是 Disposable
”
这句话 。
既然
“ 在 source.subscribe(parent)
里它是 ObservableEmitter
”
和
“ 在 observer.onSubscribe(parent)
里它是 Disposable
”
都解释清楚了，那 Observable
和 Observer
之间千丝万缕的联系也就全在上述 2 句话里了， Observable
和 Observer
的事件发布和接收就是这 2 行代码起的作用。
这里还可以解释一个问题，为什么在打印日志时我们发现
// Observer 里的 onSubscribe(@NonNull Disposable d) 方法先执行 12-15 19:05:39.665 18795-18795/com.persist.rxjava D/MainActivity: onSubscribe: isDisposable false // Observable 里的 subscribe(@NonNull ObservableEmitter<Integer> emitter) 方法后执行 12-15 19:05:39.665 18795-18795/com.persist.rxjava D/MainActivity: Observable emit 1 12-15 19:05:39.665 18795-18795/com.persist.rxjava D/MainActivity: onNext: 1 12-15 19:05:39.665 18795-18795/com.persist.rxjava D/MainActivity: Observable emit 2 12-15 19:05:39.666 18795-18795/com.persist.rxjava D/MainActivity: onNext: 2 12-15 19:05:39.666 18795-18795/com.persist.rxjava D/MainActivity: onComplete: 复制代码
因为在 subscribeActual()
方法里它们的先后顺序已经定了，注释 2 和注释 3就是它们先后顺序了，可以看下面代码。而且这也就是说 Observable
和 Observer
建立连接后， ObservableEmitter
才开始发送事件。
// 方法名暴露了真相 "实际订阅，真实订阅"
@Override
protected void subscribeActual(Observer<? super T> observer) {
// 1.创建 CreateEmitter 对象，参数传的是 observer， 这个 observer 从哪冒出来的，待会儿就知道了
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
// 2.并且这个 observer 还订阅了 CreateEmitter 对象
observer.onSubscribe(parent);
try {
// 3.source 也订阅了 CreateEmitter 对象
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
复制代码
正文到此结束
热门推荐
相关文章
近期评论
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
搞个gitee的项目
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
