引言
关于RxJava2的用法网上的资料很多，这里我们只学习它的实现原理。本文专题目的:
1.知道源头(Observable)是如何将数据发送出去的。
2.知道终点（Observer）是如何接收到数据的。
3.何时将源头和终点关联起来的
今天我们先从最简单的无背压(Observable)的create操作符说起，来解决前三个问题。
样例
//1.创建被观察者，生产事件
final Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
//2.订阅的时候发送事件
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
// emitter.onError(new Throwable("haha"));
emitter.onComplete();//onComplete事件发送后，后面的所有事件无效,且后面不能发送错误事件
emitter.onNext(3);
}
});
//3.定义观察者
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "开始采用subscribe连接");
mDisposable = d;
}
@Override
public void onNext(Integer value) {
Log.e(TAG, "对Next事件" + value + "作出响应");
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.e(TAG, "对Complete事件作出响应");
}
};
//4建立联系
observable.subscribe(observer);
我们看到出现了一下几个角色:
- Observable:被观察者，是数据的源头，通过subscribe订阅被观察者;
- ObservableOnSubscribe:从代码结构上看，Observable的构造方法需要它，且持有subscribe方法，这里暂时理解为观察者和被观察者的中间件,具体作用后面再看;
- ObservableEmitter:顾名思义，是数据发射器，被观察者通过它发送事件;
- Observer:被观察者，数据接受者，持有onNext、onError、onComplete、onSubscribe方法。
Observable
public abstract class Observable<T> implements ObservableSource<T> {
...
}
实现了ObservableSource接口:
public interface ObservableSource<T> {
/**
* Subscribes the given Observer to this ObservableSource instance.
* @param observer the Observer, not null
* @throws NullPointerException if {@code observer} is null
*/
void subscribe(Observer<? super T> observer);
}
接口很简单，提供了订阅观察者的功能，Observable中该方法的实现后面再看。我们先看看create操作符干了些啥:
public abstract class Observable<T> implements ObservableSource<T> {
...
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
//判空
ObjectHelper.requireNonNull(source, "source is null");
//构造ObservableCreate对象
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
...
}
最后通过用户构造的ObservableOnSubscribe对象，返回了ObservableCreate对象。我们先看看ObservableOnSubscribe。
数据发射封装-ObservableOnSubscribe
public interface ObservableOnSubscribe<T> {
/**
* Called for each Observer that subscribes.
* @param e the safe emitter instance, never null
* @throws Exception on error
*/
void subscribe(ObservableEmitter<T> e) throws Exception;
}
是个接口，用户发射数据就是在这个接口实现中完成，具体见样例代码，这里流一个疑问:入参ObservableEmitter是怎么来的，别急后面马上会讲到！
create操作符的产物-ObservableCreate
然后再看Observable的子类ObservableCreate，根据名字我们可以猜到它是由create操作符创建的被观察者:
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 {
//调用中间件ObservableOnSubscribe的订阅方法，开始调用发射数据的代码了！
//说明1
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
...
}
这里的source由用户构造，实现发射数据操作，subscribeActual方法是核心，当订阅观察者是，最终会执行subscribeActual方法，后面会具体说明，不过看方法名也应该能猜到。
前面我们讲ObservableOnSubscribe的subscribe方法时，关于入参的来源留下了一个疑问，这里看说明1的代码:入参parent类型为CreateEmitter，很明显它必然是ObservableEmitter的子类或者子接口。
事件发射器--ObservableEmitter
public interface ObservableEmitter<T> extends Emitter<T> {
...
}
public interface Emitter<T> {
/**
* Signal a normal value.
* @param value the value to signal, not null
*/
void onNext(T value);
/**
* Signal a Throwable exception.
* @param error the Throwable to signal, not null
*/
void onError(Throwable error);
/**
* Signal a completion.
*/
void onComplete();
}
可见发射器接口提供了发射数据的功能，在回过头来看看CreateEmitter,它是ObservableCreate的内部类.
public final class ObservableCreate<T> extends Observable<T> {
...
...
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;
}
if (!isDisposed()) {
//调用观察者的接收数据的方法
observer.onNext(t);
}
}
@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();
}
}
}
...
}
}
饶了半天，我们终于找到被观察者被调用的地方了，用户调用发射器的发送数据的方法最终会通过ObservableCreate中的CreateEmitter实现调用，而CreateEmitter最终又会调用观察者的接收数据方法，到此为止，下游接收数据的流程如下:
1.create操作符通过ObservableOnSubscribe对象构造ObservableCreate对象;
- ObservableCreate在执行订阅方法subscribeActual时，通过Observer对象构造发射器CreateEmitter;
- CreateEmitter发射数据最终会调用Observer对应的接收数据方法。
Observable的订阅方法
上面我们理解了接收数据的流程，下面我们瞅瞅Observable和Observer建立联系的订阅方法:
@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
//subscribeActual核心代码！！！
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
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
RxJavaPlugins.onError(e);
NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
npe.initCause(e);
throw npe;
}
}
lei了lei了！看到subscribeActual方法就像遇到亲人了，之前我们了解到Create操作法创建的是ObservableCreate对象，这里用户执行订阅方法时会调用subscribeActual，我们再回头看看ObservableCreate的subscribeActual实现:
@Override
protected void subscribeActual(Observer<? super T> observer) {
//根据Observer构造发射器
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);
try {
//这里由用户实现发射数据操作
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
读到这里，发现整个数据的产生和接收终于打通，订阅方法通过 source.subscribe(parent)由用户发射数据，在本样例中就是:
final Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
//2.订阅的时候发送事件
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
// emitter.onError(new Throwable("haha"));
emitter.onComplete();//onComplete事件发送后，后面的所有事件无效,且后面不能发送错误事件
emitter.onNext(3);
}
});
emitter参数就是CreateEmitter类型发射器parent。相信看到这里，整个数据的流程应该比较清晰了。数据流向如下:
Observable订阅Observer--> Observable执行subscribeActual方法--> ObservableOnSubscribe执行subscribe方法--> ObservableEmitter执行发射数据方法--> Observer执行接收数据方法。
我们可以看到，RxJava规定了数据从Observable到Observer的统一流程，至于用户发送什么数据、按什么顺序发都通过中间件ObservableOnSubscribe和ObservableEmitter实现。
