概述
Q1446595067官网:h5.haozibbs.com
最近事情太多了,现在公司内部的变动,自己岗位的变化,以及最近决定找工作。所以博客耽误了,准备面试中,打算看一看RxJava2的源码,遂有了这篇文章。
不会对RxJava2的源码逐字逐句的阅读,只寻找关键处,我们平时接触得到的那些代码。
背压实际中接触较少,故只分析了Observable
.
分析的源码版本为:2.0.1
我们的目的:
- 知道源头(
Observable
)是如何将数据发送出去的。 - 知道终点(
Observer
)是如何接收到数据的。 - 何时将源头和终点关联起来的
- 知道线程调度是怎么实现的
- 知道操作符是怎么实现的
本文先达到目的1 ,2 ,3。
我个人认为主要还是适配器模式的体现,我们接触的就只有Observable
和Observer
,其实内部有大量的中间对象在适配:将它们两联系起来,加入一些额外功能,例如考虑dispose和hook等。
从create开始。
这是一段不涉及操作符和线程切换的简单例子:
Observable.create(new ObservableOnSubscribe<String>() { @Override public void subscribe(ObservableEmitter<String> e) throws Exception { e.onNext("1"); e.onComplete(); } }).subscribe(new Observer<String>() { @Override public void onSubscribe(Disposable d) { Log.d(TAG, "onSubscribe() called with: d = [" + d + "]"); } @Override public void onNext(String value) { Log.d(TAG, "onNext() called with: value = [" + value + "]"); } @Override public void onError(Throwable e) { Log.d(TAG, "onError() called with: e = [" + e + "]"); } @Override public void onComplete() { Log.d(TAG, "onComplete() called"); } });
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
拿 create来说,
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) { //..... return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source)); }
- 1
- 2
- 3
- 4
返回值是Observable
,参数是ObservableOnSubscribe
,定义如下:
public interface ObservableOnSubscribe<T> { void subscribe(ObservableEmitter<T> e) throws Exception; }
- 1
- 2
- 3
ObservableOnSubscribe
是一个接口,里面就一个方法,也是我们实现的那个方法:
该方法的参数是 ObservableEmitter
,我认为它是关联起 Disposable
概念的一层:
public interface ObservableEmitter<T> extends Emitter<T> { void setDisposable(Disposable d); void setCancellable(Cancellable c); boolean isDisposed(); ObservableEmitter<T> serialize(); }
- 1
- 2
- 3
- 4
- 5
- 6
ObservableEmitter
也是一个接口。里面方法很多,它也继承了 Emitter<T>
接口。
public interface Emitter<T> { void onNext(T value); void onError(Throwable error); void onComplete(); }
- 1
- 2
- 3
- 4
- 5
Emitter<T>
定义了 我们在ObservableOnSubscribe
中实现subscribe()
方法里最常用的三个方法。
好,我们回到原点,create()
方法里就一句话return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
,其中提到RxJavaPlugins.onAssembly()
:
/** * Calls the associated hook function. * @param <T> the value type * @param source the hook's input value * @return the value returned by the hook */ @SuppressWarnings({ "rawtypes", "unchecked" }) public static <T> Observable<T> onAssembly(Observable<T> source) { Function<Observable, Observable> f = onObservableAssembly; if (f != null) { return apply(f, source); } return source; }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
可以看到这是一个关于hook的方法,关于hook我们暂且不表,不影响主流程,我们默认使用中都没有hook,所以这里就是直接返回source
,即传入的对象,也就是new ObservableCreate<T>(source)
.
ObservableCreate
我认为算是一种适配器的体现,create()
需要返回的是Observable
,而我现在有的是(方法传入的是)ObservableOnSubscribe
对象,ObservableCreate
将ObservableOnSubscribe
适配成Observable
。
其中subscribeActual()
方法表示的是被订阅时真正被执行的方法,放后面解析:
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) { CreateEmitter<T> parent = new CreateEmitter<T>(observer); observer.onSubscribe(parent); try { source.subscribe(parent); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); parent.onError(ex); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
OK,至此,创建流程结束,我们得到了Observable<T>
对象,其实就是ObservableCreate<T>
.
到订阅subscribe 结束
subscribe()
:
public final void subscribe(Observer<? super T> observer) { ... try { //1 hook相关,略过 observer = RxJavaPlugins.onSubscribe(this, observer); ... //2 真正的订阅处 subscribeActual(observer); } catch (NullPointerException e) { // NOPMD throw e; } catch (Throwable e) { //3 错误处理, Exceptions.throwIfFatal(e); // can't call onError because no way to know if a Disposable has been set or not // can't call onSubscribe because the call might have set a Subscription already //4 hook错误相关,略过 RxJavaPlugins.onError(e); NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS"); npe.initCause(e); throw npe; } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
关于hook的代码:
可以看到如果没有hook,即相应的对象是null,则是传入什么返回什么的。
/** * Calls the associated hook function. * @param <T> the value type * @param source the hook's input value * @param observer the observer * @return the value returned by the hook */ @SuppressWarnings({ "rawtypes", "unchecked" }) public static <T> Observer<? super T> onSubscribe(Observable<T> source, Observer<? super T> observer) { //1 默认onObservableSubscribe(可理解为一个flatmap的操作)是null BiFunction<Observable, Observer, Observer> f = onObservableSubscribe; //2 所以这句跳过,不会对其进行apply if (f != null) { return apply(f, source, observer); } //3 返回参数2 return observer; }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
我也是验证了一下 三个Hook相关的变量,确实是null:
Consumer<Throwable> errorHandler = RxJavaPlugins.getErrorHandler(); BiFunction<Observable, Observer, Observer> onObservableSubscribe = RxJavaPlugins.getOnObservableSubscribe(); Function<Observable, Observable> onObservableAssembly = RxJavaPlugins.getOnObservableAssembly(); Log.e(TAG, "errorHandler = [" + errorHandler + "]"); Log.e(TAG, "onObservableSubscribe = [" + onObservableSubscribe + "]"); Log.e(TAG, "onObservableAssembly = [" + onObservableAssembly + "]");
- 1
- 2
- 3
- 4
- 5
- 6
- 7
所以订阅时的重点就是:
//2 真正的订阅处 subscribeActual(observer);
- 1
- 2
我们将第一节提到的ObservableCreate
里的subscribeActual()
方法拿出来看看:
@Override protected void subscribeActual(Observer<? super T> observer) { //1 创建CreateEmitter,也是一个适配器 CreateEmitter<T> parent = new CreateEmitter<T>(observer); //2 onSubscribe()参数是Disposable ,所以CreateEmitter可以将Observer->Disposable 。还有一点要注意的是`onSubscribe()`是在我们执行`subscribe()`这句代码的那个线程回调的,并不受线程调度影响。 observer.onSubscribe(parent); try { //3 将ObservableOnSubscribe(源头)与CreateEmitter(Observer,终点)联系起来 source.subscribe(parent); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); //4 错误回调 parent.onError(ex); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
Observer
是一个接口,里面就四个方法,我们在开头的例子中已经全部实现(打印Log)。
public interface Observer<T> { void onSubscribe(Disposable d); void onNext(T value); void onError(Throwable e); void onComplete(); }
- 1
- 2
- 3
- 4
- 5
- 6
重点在这一句:
//3 将ObservableOnSubscribe(源头)与CreateEmitter(Observer,终点)联系起来 source.subscribe(parent);
- 1
- 2
source
即ObservableOnSubscribe
对象,在本文中是:
new ObservableOnSubscribe<String>() { @Override public void subscribe(ObservableEmitter<String> e) throws Exception { e.onNext("1"); e.onComplete(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
则会调用parent.onNext()
和parent.onComplete()
,parent
是CreateEmitter
对象,如下:
static final class CreateEmitter<T> extends AtomicReference<Disposable> implements ObservableEmitter<T>, Disposable { final Observer<? super T> observer; CreateEmitter(Observer<? super T> observer) { this.observer = observer; } @Override public void onNext(T t) { ... //如果没有被dispose,会调用Observer的onNext()方法 if (!isDisposed()) { observer.onNext(t); } } @Override public void onError(Throwable t) { ... //1 如果没有被dispose,会调用Observer的onError()方法 if (!isDisposed()) { try { observer.onError(t); } finally { //2 一定会自动dispose() dispose(); } } else { //3 如果已经被dispose了,会抛出异常。所以onError、onComplete彼此互斥,只能被调用一次 RxJavaPlugins.onError(t); } } @Override public void onComplete() { //1 如果没有被dispose,会调用Observer的onComplete()方法 if (!isDisposed()) { try { observer.onComplete(); } finally { //2 一定会自动dispose() dispose(); } } } @Override public void dispose() { DisposableHelper.dispose(this); } @Override public boolean isDisposed() { return DisposableHelper.isDisposed(get()); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
总结重点:
-
Observable
和Observer
的关系没有被dispose
,才会回调Observer
的onXXXX()
方法 -
Observer
的onComplete()
和onError()
互斥只能执行一次,因为CreateEmitter
在回调他们两中任意一个后,都会自动dispose()
。根据第一点,验证此结论。 -
Observable
和Observer
关联时(订阅时),Observable
才会开始发送数据。 -
ObservableCreate
将ObservableOnSubscribe
(真正的源)->Observable
. -
ObservableOnSubscribe
(真正的源)需要的是发射器ObservableEmitter
. -
CreateEmitter
将Observer
->ObservableEmitter
,同时它也是Disposable
. - 先
error
后complete
,complete
不显示。 反之会crash,感兴趣的可以写如下代码验证。
e.onNext("1"); //先error后complete,complete不显示。 反之 会crash //e.onError(new IOException("sb error")); e.onComplete(); e.onError(new IOException("sb error"));
- 1
- 2
- 3
- 4
- 5
一个好玩的地方DisposableHelper
原本到这里,最简单的一个流程我们算是搞清了。
还值得一提的是,DisposableHelper.dispose(this);
DisposableHelper
很有趣,它是一个枚举,这是利用枚举实现了一个单例disposed state
,即是否disposed,如果Disposable
类型的变量的引用等于DISPOSED
,则起点和终点已经断开联系。
其中大多数方法 都是静态方法,所以isDisposed()
方法的实现就很简单,直接比较引用即可.
其他的几个方法,和AtomicReference
类搅基在了一起。
这是一个实现引用原子操作的类,对象引用的原子更新
,常用方法如下:
//返回当前的引用。 V get() //如果当前值与给定的expect引用相等,(注意是引用相等而不是equals()相等),更新为指定的update值。 boolean compareAndSet(V expect, V update) //原子地设为给定值并返回旧值。 V getAndSet(V newValue)
- 1
- 2
- 3
- 4
- 5
- 6
OK,铺垫完了我们看看源码吧:
public enum DisposableHelper implements Disposable { /** * The singleton instance representing a terminal, disposed state, don't leak it. */ DISPOSED ; public static boolean isDisposed(Disposable d) { return d == DISPOSED; } public static boolean dispose(AtomicReference<Disposable> field) { //1 通过断点查看,默认情况下,field的值是"null",并非引用是null哦!大坑大坑大坑 //但是current是null引用 Disposable current = field.get(); Disposable d = DISPOSED; //2 null不等于DISPOSED if (current != d) { //3 field是DISPOSED了,current还是null current = field.getAndSet(d); if (current != d) { //4 默认情况下 走不到这里,这里是在设置了setCancellable()后会走到。 if (current != null) { current.dispose(); } return true; } } return false; }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
总结
- 在
subscribeActual()
方法中,源头和终点关联起来。 -
source.subscribe(parent);
这句代码执行时,才开始从发送ObservableOnSubscribe
中利用ObservableEmitter
发送数据给Observer
。即数据是从源头push给终点的。 -
CreateEmitter
中,只有Observable
和Observer
的关系没有被dispose
,才会回调Observer
的onXXXX()
方法 -
Observer
的onComplete()
和onError()
互斥只能执行一次,因为CreateEmitter
在回调他们两中任意一个后,都会自动dispose()
。根据上一点,验证此结论。 - 先
error
后complete
,complete
不显示。 反之会crash - 还有一点要注意的是
onSubscribe()
是在我们执行subscribe()
这句代码的那个线程回调的,并不受线程调度影响。