RxJava 操作符的原理

今天再一次看Rxjava的几个操作符时发现对于操作符到底做了什么事不是很清楚，使用just，create等操作符创建一个Observable，和使用filter、map等操作符对Observable发送的数据进行转换有什么区别和联系？filter和map这样的操作符最终是如何对数据进行处理的？带着这几个问题再次对对应的源码进行查看并做以下记录。

Just、create等创建操作符：

首先看下just操作符的源码，做了什么：

Observable.just("","").subscribe({/// 观察者对象代码...
})

   public static <T> Observable<T> just(T item1, T item2) {ObjectHelper.requireNonNull(item1, "The first item is null");ObjectHelper.requireNonNull(item2, "The second item is null");return fromArray(item1, item2);}

 public static <T> Observable<T> fromArray(T... items) {ObjectHelper.requireNonNull(items, "items is null");if (items.length == 0) {return empty();} elseif (items.length == 1) {return just(items[0]);}return RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items));}

just操作符调用了一个多参数的just方法，传入创建参数，然后调用fromArray方法处理，

最终调用RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items))组装了一个对象返回。

public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {Function<? super Observable, ? extends Observable> f = onObservableAssembly;if (f != null) {return apply(f, source);}return source;}

onAssembly() 方法在满足内部字段不为null的时候会调用一次apply方法处理，否则则直接把创建传入的Observable返回。最后在调用Observable的订阅方法subscribe的时候执行实现类的subscribeActual方法逻辑。

实现类ObservableFromArray来处理传入的数组类型数据。

//继承了 Observable 抽象类
public final class ObservableFromArray<T> extends Observable<T> {final T[] array;public ObservableFromArray(T[] array) {this.array = array;}@Overridepublic void subscribeActual(Observer<? super T> s) {// 内部 通过 FromArrayDisposable来处理数据FromArrayDisposable<T> d = new FromArrayDisposable<T>(s, array);s.onSubscribe(d);if (d.fusionMode) {return;}// 调用数据处理的方法d.run();}static final class FromArrayDisposable<T> extends BasicQueueDisposable<T> {//订阅的时候传入的观察者对象final Observer<? super T> actual;final T[] array;int index;boolean fusionMode;volatile boolean disposed;FromArrayDisposable(Observer<? super T> actual, T[] array) {this.actual = actual;this.array = array;}@Overridepublic int requestFusion(int mode) {if ((mode & SYNC) != 0) {fusionMode = true;return SYNC;}return NONE;}@Nullable@Overridepublic T poll() {int i = index;T[] a = array;if (i != a.length) {index = i + 1;return ObjectHelper.requireNonNull(a[i], "The array element is null");}return null;}@Overridepublic boolean isEmpty() {return index == array.length;}@Overridepublic void clear() {index = array.length;}@Overridepublic void dispose() {disposed = true;}@Overridepublic boolean isDisposed() {return disposed;}void run() {T[] a = array;int n = a.length;// 遍历数组，不断的发送传入的数组中的元素调用onNext方法for (int i = 0; i < n && !isDisposed(); i++) {T value = a[i];if (value == null) {actual.onError(new NullPointerException("The " + i + "th element is null"));return;}actual.onNext(value);}if (!isDisposed()) {actual.onComplete();}}}
}

ObservableFromArray 类中的 subscribeActual方法，调用FromArrayDisposable进行处理。

在看下调用subscribe方法订阅的时候做了什么：

 //public abstract class Observable<T> 类中的方法 subscribepublic 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这个方法就在操作符的// 实现类中实现，比如just操作符的实现类 ObservableFromArraysubscribeActual(observer);} catch (NullPointerException e) { // NOPMDthrow 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 alreadyRxJavaPlugins.onError(e);NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");npe.initCause(e);throw npe;}}

可以看到我们订阅的时候调用了 Observable中的方法subscribe，里边调用了其抽象方法subscribeActual(observer);传入了观察者对象。而 subscribeActual方法的具体实现就在各个操作符的实现类中。比如上面说的just操作符的实现类ObservableFromArray，实现了just的具体逻辑。

take 、filter、map等转换操作符是如何实现的

// take和filter操作符，是如何实现的？
sub.take(1)
sub.filter { 1==1 }

public final Observable<T> take(long count) {if (count < 0) {throw new IllegalArgumentException("count >= 0 required but it was " + count);}//take 操作符最后创建了 ObservableTake对象return RxJavaPlugins.onAssembly(new ObservableTake<T>(this, count));}

public final Observable<T> filter(Predicate<? super T> predicate) {ObjectHelper.requireNonNull(predicate, "predicate is null");// filter 操作符，创建了ObservableFilter对象return RxJavaPlugins.onAssembly(new ObservableFilter<T>(this, predicate));}

以上两个操作符最后都使用了RxJavaPlugins这个类的onAssembly方法进行组装。各自创建了一个实现类的对象。

// ObservableTake类
public final class ObservableTake<T> extends AbstractObservableWithUpstream<T, T> {final long limit;public ObservableTake(ObservableSource<T> source, long limit) {super(source);this.limit = limit;}@Overrideprotected void subscribeActual(Observer<? super T> observer) {source.subscribe(new TakeObserver<T>(observer, limit));}static final class TakeObserver<T> implements Observer<T>, Disposable {final Observer<? super T> actual;boolean done;Disposable subscription;long remaining;TakeObserver(Observer<? super T> actual, long limit) {this.actual = actual;this.remaining = limit;}@Overridepublic void onSubscribe(Disposable s) {if (DisposableHelper.validate(this.subscription, s)) {subscription = s;if (remaining == 0) {done = true;s.dispose();EmptyDisposable.complete(actual);} else {actual.onSubscribe(this);}}}@Overridepublic void onNext(T t) {// 处理take的参数，不为0的时候就onNext方法中就自减1，否则才调用onNext传数据，if (!done && remaining-- > 0) {boolean stop = remaining == 0;actual.onNext(t);if (stop) {onComplete();}}}@Overridepublic void onError(Throwable t) {if (done) {RxJavaPlugins.onError(t);return;}done = true;subscription.dispose();actual.onError(t);}@Overridepublic void onComplete() {if (!done) {done = true;subscription.dispose();actual.onComplete();}}@Overridepublic void dispose() {subscription.dispose();}@Overridepublic boolean isDisposed() {return subscription.isDisposed();}}
}

可以看到ObservableTake处理了take操作符参数的逻辑，在onNext中判断参数是否是0，不为0则不断的自减，直到0才调用onNext方法发送数据。

//ObservableFilter filter操作符实现
public final class ObservableFilter<T> extends AbstractObservableWithUpstream<T, T> {final Predicate<? super T> predicate;public ObservableFilter(ObservableSource<T> source, Predicate<? super T> predicate) {super(source);this.predicate = predicate;}@Overridepublic void subscribeActual(Observer<? super T> s) {source.subscribe(new FilterObserver<T>(s, predicate));}static final class FilterObserver<T> extends BasicFuseableObserver<T, T> {final Predicate<? super T> filter;FilterObserver(Observer<? super T> actual, Predicate<? super T> filter) {super(actual);this.filter = filter;}@Overridepublic void onNext(T t) {if (sourceMode == NONE) {boolean b;try {//调用filter操作符的方法获取一个boolean的值                    b = filter.test(t);} catch (Throwable e) {fail(e);return;}if (b) {// 满足条件才调用 onNext方法actual.onNext(t);}} else {actual.onNext(null);}}@Overridepublic int requestFusion(int mode) {return transitiveBoundaryFusion(mode);}@Nullable@Overridepublic T poll() throws Exception {for (;;) {T v = qs.poll();if (v == null || filter.test(v)) {return v;}}}}
}

同样的filter操作符的实现中，通过获取filter的值，进行了判断，满足条件的才调用onNext发送数据。

如果我们对一个数据调用了多个操作符处理：

// 连续调用 just 创建，然后调用take和map，最后使用subscribe订阅
Observable.just(1,5,6,7,8,10,).filter { it%2 == 0 }.map {println("map == : $it")it * 2}.subscribe {println("subscribe -result== : $it")}

从上面的分析实际上可以知道：每个操作符都返回了一个Observable对象：

由上一个操作符返回传给下一个操作符继续调用下一个操作符的实现逻辑。每个满足的数据都是一个个的进行处理，最终交给订阅者接收。

所以，本质上操作符就是我们对原始创建的Observable对象做的一次转换，每个操作符的实现都是Observable抽象类的一个子类，中间通过对操作符逻辑实现转换，调用观察者对象的onNext方法发送数据，满足操作符的条件就发送，否则做对应的处理（过滤、转换、取值等）。

参考文献

RxJava(10-操作符原理&自定义操作符)