前语
重温RxJava2源码，做个简单的记载，本文仅剖析事情的发射与消费简单逻辑，从源码视点剖析被观察者（上游事情）是怎么与观察者（下流事情）进行相关的。
事情发射
Observable.just(1,2,3)
.subscribe();
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}).subscribe();
上述两种办法都是由被观察者发出3个事情，交给观察者（下流事情）去处理。这儿剖析一下Observable.just 与Observable.create 办法的差异
Observable被观察者（上游事情）
just办法
public static <T> Observable<T> just(T item1, T item2, T item3) {
return fromArray(item1, item2, item3);
}
这儿将传入的item…持续传入fromArray办法
public static <T> Observable<T> fromArray(T... items) {
return RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items));
}
终究将参数传入实例化的ObservableFromArray目标中，并将该目标回来，此处可先不关注RxJavaPlugins类，持续探索ObservableFromArray类都做了什么；
public final class ObservableFromArray<T> extends Observable<T> {
final T[] array;
public ObservableFromArray(T[] array) {
this.array = array;
}
@Override
public void subscribeActual(Observer<? super T> observer) {
FromArrayDisposable<T> d = new FromArrayDisposable<T>(observer, array);
observer.onSubscribe(d);
if (d.fusionMode) {
return;
}
d.run();
}
}
作为Observable的子类，每个被观察者都要完成自己的subscribeActual办法，这儿才是真正与观察者进行绑定的具体完成，其间实例化了FromArrayDisposable目标，并将observer（观察者）与array传入，办法完毕调用了其run办法。
void run() {
T[] a = array;
int n = a.length;
for (int i = 0; i < n && !isDisposed(); i++) {
T value = a[i];
if (value == null) {
downstream.onError(new NullPointerException("The element at index " + i + " is null"));
return;
}
downstream.onNext(value);
}
if (!isDisposed()) {
downstream.onComplete();
}
}
能够看到其间对于开始传入的1、2、3，以此进行了onNext办法的调用，分发完毕后调用了onComplete，事情完毕。
create办法
首先从上面的实例代码能够看到，create办法中还需求传入ObservableOnSubscribe的实例目标，暂时不管，咱们来挖掘一下create办法
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
终究将上述咱们创建的ObservableOnSubscribe目标传入新实例化的ObservableCreate目标中，并将该目标回来；
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);
}
}
}
看到在subscribeActual办法中，创建了CreateEmitter目标，接着别离调用observer#onSubscribe办法和source#subscribe办法，这儿要搞清楚其间的3个变量别离是什么
-
source：被观察者（上游事情），开始咱们create办法中传入的接口目标，咱们便是在source中进行事情分发的 -
observer：观察者（下流事情），咱们的事情终究交给observer去处理，这儿将observer传入了CreateEmitter，便是要在Emitter中进行中转分发事情给observer -
parent：理解为一个上下流的中转站，上游事情发射后在这儿交给下流去处理
终究咱们看一下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;
}
if (!isDisposed()) {
observer.onNext(t);
}
}
}
这儿只贴出了onNext办法，能够看到当onNext办法被调用后，其间就会去调用observer的onNext办法，而onNext开始的触发便是在实例代码中咱们实例化的ObservableOnSubscribe其间的subscribe办法中
事情消费
...
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
}
});
...
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull Integer integer) {
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
}
});
上述两种办法都是接纳被观察者（上游事情）发出的事情，进行处理消费。这儿剖析一下Consumer与Observer的差异
Observer观察者（下流事情）
Consumer
public interface Consumer<T> {
/**
* Consume the given value.
* @param t the value
* @throws Exception on error
*/
void accept(T t) throws Exception;
}
Consumer仅为一个接口类，其间accept办法接纳事情并消费，咱们需求去到上游事情订阅下流事情时的subscribe办法，依据下流事情的参数类型与数量，会进入不同的subscribe重载办法中；
subscribe(Consumer<? super T> onNext) : Diposable
public final Disposable subscribe(Consumer<? super T> onNext) {
return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
Action onComplete, Consumer<? super Disposable> onSubscribe) {
LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
subscribe(ls);
return ls;
}
该办法中包装了一个LambdaObserver，将咱们传入的onNext办法再传入其间
public final class LambdaObserver<T> extends AtomicReference<Disposable>
implements Observer<T>, Disposable, LambdaConsumerIntrospection {
private static final long serialVersionUID = -7251123623727029452L;
final Consumer<? super T> onNext;
final Consumer<? super Throwable> onError;
final Action onComplete;
final Consumer<? super Disposable> onSubscribe;
public LambdaObserver(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
Action onComplete,
Consumer<? super Disposable> onSubscribe) {
super();
this.onNext = onNext;
this.onError = onError;
this.onComplete = onComplete;
this.onSubscribe = onSubscribe;
}
@Override
public void onNext(T t) {
if (!isDisposed()) {
try {
onNext.accept(t);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
get().dispose();
onError(e);
}
}
}
能够看到LambdaObserver实际上便是Observer的完成类，其间完成了onSubscribe onNext onError onComplete 办法，上述代码中咱们看到咱们开始的Consumer目标实际上便是其间的onNext变量，在LambdaObserver收到onNext事情消费时，再将事情交给Consumer去处理。Consumer相当于一种简易形式的观察者，依据被观察者的subscribe订阅办法消费特定的事情（onNext或onError等）。
Observer
public interface Observer<T> {
void onSubscribe(@NonNull Disposable d);
void onNext(@NonNull T t);
void onError(@NonNull Throwable e);
void onComplete();
}
Observer是最原始的观察者，是一切Observer的顶层接口，其间办法为观察者能够消费的四个事情
subscribe(Observer<? super T> observer)
该办法也是其他一切订阅观察者办法终究会进入的办法
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
...
} catch (Throwable e) {
...
}
}
终究在subscribeActual办法中进行被观察者与观察者（上游与下流事情）的绑定。
写在完毕
抛开一切的操作符、线程切换来说，RxJava的上下流事情绑定逻辑还是十分清晰易读的，能够通过源码了解每个事情是怎么从上游传递至下流的。至于其他逻辑，另起篇幅剖析。
