RxJava2 事件分发&消费绑定逻辑简析-六虎

前语

重温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的上下流事情绑定逻辑还是十分清晰易读的，能够通过源码了解每个事情是怎么从上游传递至下流的。至于其他逻辑，另起篇幅剖析。

RxJava2 事件分发&消费绑定逻辑 简析

前语

事情发射

Observable被观察者（上游事情）

just办法

create办法

事情消费

Observer观察者（下流事情）

Consumer

Observer

写在完毕

相关文章

SwiftUI集成百度地图

ExoPlayer架构详解与源码分析（13）——TeeDataSource和CacheDataSource

Java并发编程之Unsafe

ANR如何产生之InputDispatching Timeout篇

作者信息

RxJava2 事件分发&消费绑定逻辑简析