RxJava

---

前言

Linus Benedict Torvalds : RTFSC – Read The Fucking Source Code

概括

原理解析我们从三个部分入手：1、基本流程梳理，2、lift的实现原理，3、线程切换原理。

基本流程

Observable.create(new Observable.OnSubscribe<ArrayList<String>>() {
              @Override
              public void call(Subscriber<? super ArrayList<String>> subscriber) {
                  subscriber.onNext(T);
                  subscriber.onCompleted();
              }
          })
          .subscribe(new Subscriber<String>() {
              @Override
              public void onCompleted() {}
              @Override
              public void onError(Throwable e) {}
              @Override
              public void onNext(String s) {}
          });

这就是Rxjava的基本使用流程。我们从流程开始分析：

Observable.create()

// Step 1:
public final static <T> Observable<T> create(OnSubscribe<T> f) {
    return new Observable<T>(hook.onCreate(f));
}

// Step 2:
protected Observable(OnSubscribe<T> f) {
    this.onSubscribe = f;
}

这两步是创建了一个Observable对象和OnSubscribe对象。

Observable.subscribe()

private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
    // new Subscriber so onStart it
    subscriber.onStart();
    
    // allow the hook to intercept and/or decorate
    hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
}

这一步的操作实际上就是调用了 Observable.onSubscribe.call(subscriber);

整理下基本流程分两步：
1、创建Observable, OnSubscribe, Subscriber 三个对象
2、调用Observable的subscribe方法

那么他们最终的程序调用顺序是：
1、Observable.subscribe()
2、Observable.OnSubscribe.call()
3、Subscriber.onNext()

map操作符和lift原理

RxJava之所以好用是因为它提供了很多非常简洁的操作符动作，利用lift原理可以很流畅的实现操作的顺序过度。

Observable.create(new Observable.OnSubscribe<ArrayList<String>>() {
              @Override
              public void call(Subscriber<? super ArrayList<String>> subscriber) {
                  subscriber.onNext(T);
                  subscriber.onCompleted();
              }
          })
          .map(new Func1<String, String>() {
              @Override
              public String call(String s) {
                  return null;
              }
          })
          .subscribe(new Subscriber<String>() {
              @Override
              public void onCompleted() {}
              @Override
              public void onError(Throwable e) {}
              @Override
              public void onNext(String s) {}
          });

Observable.map()

public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
    return lift(new OperatorMap<T, R>(func));
}

其实可以看到我们的map函数的主要实现是调用了litf方法，并在里面传入的是OperatorMap的对象。

Observable lift()

public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
    return new Observable<R>(new OnSubscribe<R>() {
        @Override
        public void call(Subscriber<? super R> o) {
            Subscriber<? super T> st = hook.onLift(operator).call(o);
            st.onStart();
            onSubscribe.call(st);
        }
    });
}

1. 首先 new 了一个 Func函数, 保存在了 OperatorMap.transform 中。
2. new了一个新的 Observable. 这个 Observable 的构造函数中, 传入了一个新的 OnSubscribe。
   lift的重点就在这个OnSubscribe里面。

先来看下OperatorMap的代码(只列出核心)：

public final class OperatorMap<T, R> implements Operator<R, T> {
    @Override
    public Subscriber<? super T> call(final Subscriber<? super R> o) {
        return new Subscriber<T>(o) {
            @Override
            public void onNext(T t) {
                o.onNext(transformer.call(t));
            }
        };
    }
}

我们看到从hook.onLift(operator).call(o);这里创建出来的Subscriber的call函数里主要做的就是transformer.call(t)也就是转换的操作。然后再将其转换的结果返回给onNext。

lift 流程总结

总结下整个回调流程：

1. 调用Create函数，创建一个新的Observable，并且创建一个OnSubscribe。
2. 调用Map函数，再创建一个新的Observable，同时也创建一个新的OnSubscribe。并且其中的OnSubscribe.call()方法调用了第一步的Observable.onSubscribe.call方法，最后将结果进行transform 操作。
3. 最后将拿到的transform 操作结果传给我们最后的onNext()方法中作为结果。

subscribeOn()和observeOn()线程切换原理

Rxjava还有另外一个原因让它在使用上非常简洁，就是线程处理上。
首先来看下他们的作用域

Observable.just(1, 2, 3, 4) // IO 线程，由 subscribeOn() 指定
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.newThread())
    .map(mapOperator) // 新线程，由 observeOn() 指定
    .observeOn(Schedulers.io())
    .map(mapOperator2) // IO 线程，由 observeOn() 指定
    .observeOn(AndroidSchedulers.mainThread) 
    .subscribe(subscriber);  // Android 主线程，由 observeOn() 指定

我们先上一个例子：

Observable.create(new Observable.OnSubscribe<ArrayList<String>>() {
              @Override
              public void call(Subscriber<? super ArrayList<String>> subscriber) {
                  subscriber.onNext(T);
                  subscriber.onCompleted();
              }
          })
          .subscribeOn(Schedulers.io())
          .observeOn(Schedulers.newThread())
          .subscribe(new Subscriber<String>() {
              @Override
              public void onCompleted() {}
              @Override
              public void onError(Throwable e) {}
              @Override
              public void onNext(String s) {}
          });

首先来看下observeOn()代码：

public final Observable<T> observeOn(Scheduler scheduler) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
    }
    return lift(new OperatorObserveOn<T>(scheduler));
}
public Subscriber<? super T> call(Subscriber<? super T> child) {
    if (scheduler instanceof ImmediateScheduler) {
        return child;
    } else if (scheduler instanceof TrampolineScheduler) {
        return child;
    } else {
        ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler, child);
        parent.init();
        return parent;
    }
}

可以看到observeOn其实也是调用了OperatorObserveOn类来实现的，本意上所有的Rxjava都是以操作符的形式来调用。接下来就是OperatorObserveOn的call()方法，该方法回根据传入的Subscriber类型决定返回的方式。
总结来说就是通过lift()方法将接下来传入执行的subscribe方法通过新的线程方式执行，并且在结束后再接着往下走。

总结

Rxjava 的原理我们并没有每条都往里深入，因为rxjava的代码复杂度很高，需要跟着代码有耐心的往下走才能看清，建议读者可以根据源码一步步跟进。