RxJava2.0操作符
2020-05-31 本文已影响0人
only_one
前言
RxJava这个词对于很多开发的小伙伴来说早已不知听说过多少遍，如果有不太了解的，推荐RxJava 入门 抛物线写的会让你对RxJava，有个认识，本文记录map，create操作符的原理，其他的操作都是一样的。
image.png
RxJava操作符太多了，记不住啊，就举一两个来说。
一、操作符
create
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("item1");
subscriber.onNext("item2");
subscriber.onCompleted();
}
});
map
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onCompleted();
}
}).map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return integer+"";
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e("TAG",s);
}
});
Observable（上游），Observer（下游）从上面代码片段可以看出上游往下游发送两个int的数字，通过map进行拦截装饰将之前的int类型的数字转成String类型，最终发送给下游,如图所示。
image.png
为什么可以这样操作？
首先看一下map操作符中做了那些操作
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
不难看出在调用map方法的时候new 一个ObservableMap 其实在RxJava中是一一对应的，我们看看其他操作符，一个不具备说服力
image.png
可以得出
image.png
下面我们分析整个执行的流程：
1、从Observer开始：
public interface Observer<T> {
void onSubscribe(@NonNull Disposable d);
void onNext(@NonNull T t);
void onError(@NonNull Throwable e);
void onComplete();
}
提供了onSubscribe，onNext，onError，onComplete四个方法。
2、 .subscribe(Observer 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(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;
}
}
在subscribe()方法中调用subscribeActual(observer)，点击subscribeActual方法追溯到Observable类，从上面map与ObservableMap是一一对应，那么在这儿可以得出结论，ObservableMap是Observable的子类，对不对我说了不算瞅瞅源码：
public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
abstract class AbstractObservableWithUpstream<T, U> extends Observable<U> implements HasUpstreamObservableSource<T> {
ObservableMap类中做了那些操作：
public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
super(source);
this.function = function;
}
@Override
public void subscribeActual(Observer<? super U> t) {
source.subscribe(new MapObserver<T, U>(t, function));
}
subscribeActual上面已经分析，是Observable提供的抽象方法，上面留下两个疑问，source从何而来，MapObserve是干啥的
what???.png
MapObserver类瞅瞅：
static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;
MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
super(actual);
this.mapper = mapper;
}
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != NONE) {
actual.onNext(null);
return;
}
U v;
try {
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
actual.onNext(v);
}
对Observer在MapObserver进行一个封装、管理，带着source这个问题继续往下看。
Observable.create()中传ObservableOnSubscribe（自定义资源）查看create()源码:
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
照旧瞅瞅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);
}
}
当看到source对象的时候，是不是刚刚的疑问已经解开了
直接上CreateEmitter类，代码片段
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 (!tryOnError(t)) {
RxJavaPlugins.onError(t);
}
}
@Override
public boolean tryOnError(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();
}
return true;
}
return false;
}
@Override
public void onComplete() {
if (!isDisposed()) {
try {
observer.onComplete();
} finally {
dispose();
}
}
}
通过调用CreateEmitter中的onNext()方法调用observer.onNext(t);然后通过层层调用，ObserverMap中的OnNext()将事件传递，最终传递到下游。
二、总结
整个大致的执行流程如图：倒着写的U型
image.png
时序图
image.png
