RxJava2 源码浅析

RxJava2 源码浅析

ReactiveX

历史：
ReactiveX是Reactive Extensions的缩写，一般简写为Rx，最初是LINQ的一个扩展，由微软的架构师Erik Meijer领导的团队开发，在2012年11月开源，Rx是一个编程模型，目标是提供一致的编程接口，帮助开发者更方便的处理异步数据流，Rx库支持.NET、JavaScript和C++，Rx近几年越来越流行了，现在已经支持几乎全部的流行编程语言了，Rx的大部分语言库由ReactiveX这个组织负责维护，比较流行的有RxJava/RxJS/Rx.NET，社区网站是 reactivex.io

定义：
ReactiveX.io给的定义是，Rx是一个使用可观察数据流进行异步编程的编程接口，ReactiveX结合了观察者模式、迭代器模式和函数式编程的精华。
ReactiveX不仅仅是一个编程接口，它是一种编程思想的突破，它影响了许多其它的程序库和框架以及编程语言。

filter操作符

这就是数据流？

RxJava2定义

a library for composing asynchronous and event-based programs by using observable sequences.(一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）

大致流程

上代码：

Observable.
            create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                    emitter.onNext("test First");
                    Log.e("TAG", "subScribe test First");
                    emitter.onNext("test Second");
                    Log.e("TAG", "subScribe test Second");
                    emitter.onComplete();
                    Log.e("TAG", "subScribe onComplete");
                }
            }).
            subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.e("TAG", "onSubscribe");
                    mDisposable = d;
                }

                @Override
                public void onNext(String s) {
                    Log.e("TAG", "onNext");
                    Log.e("TAG", s);
                    if (s.equals("test First")) {
                        mDisposable.dispose();
                    }
                }

                @Override
                public void onError(Throwable e) {
                    Log.e("TAG", "onError");
                }

                @Override
                public void onComplete() {
                    Log.e("TAG", "onComplete");
                }
            });

刚开始学习RxJava 时，这段代码给我最直观的感受就是，这不就是自己调用自己吗。ObservableEmitter<String> emitter 这个就是下面的subscribe(new Observer<String>()）。对吧，我觉得大家应该都是这样的感受吧...

追踪一下源码：点击create()方法进去看一下：

@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));
}


public static <T> Observable<T> onAssembly(Observable<T> source) {
    Function<Observable, Observable> f = onObservableAssembly;
    if (f != null) {
        return apply(f, source);
    }
    return source;
}
看了代码onObservableAssembly为null,所以create方法之后这个对象被包装成new ObservableCreate<T>(source)，source是外面传进来的。

关键字：io.reactivex.internal.operators.observable.ObservableCreate

继续看下一个操作符：subscribe()，点进去看一下

@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        observer = RxJavaPlugins.onSubscribe(this, observer);//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;
    }
}
// RxJavaPlugins.onSubscribe

  public static <T> Observer<? super T> onSubscribe(Observable<T> source, Observer<? super T> observer) {
    BiFunction<Observable, Observer, Observer> f = onObservableSubscribe;   //f 为null
    if (f != null) {
        return apply(f, source, observer);
    }
    return observer;
}

上面说过了经过create()方法或这个对象已经是ObservableCreate了，那么最终会调用的就是subscribeActual(observer) 看一下ObservableCreate这个类的代码：

@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);
    }
}


static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
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);
        }
    }
}

看一下，这里会把observer包装成一个CreateEmitter对象，然后source是Observable.create(new ObservableOnSubscribe<String>())传进来的ObservableOnSubscribe对象。然后会调用observer.onSubscribe(parent);source.subscribe(parent);终于清晰了...可以回答上面的问题了，其实Observer和ObservableEmitter可以看成是一个对象，只是对observer做了个包装...

Scheduler 线程变换(subscribeOn 和 observeOn)

说到线程变换即线程间通信，因为我是学Android，所以第一印象就是Handler，然后就是Future。看了源码后发现RxJava用的是Future，ScheduledExecutorService，Runnable，二AndroidScheduler就是用Handler的，因为需要切换到Android中的UI线程。

subscribeOn(Schedulers.newThread())

点进去看一下：

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;

public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
    super(source);
    this.scheduler = scheduler;
}

@Override
public void subscribeActual(final Observer<? super T> s) {
    final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

    s.onSubscribe(parent);

    parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
        @Override
        public void run() {
            source.subscribe(parent);
        }
    }));
}
}

传进来的是Schedulers.newThread()，点击Schedulers.newThread() 点进去
发现最终返回的是NewThreadScheduler

关键字：io.reactivex.internal.schedulers.NewThreadScheduler
io.reactivex.internal.schedulers.NewThreadWorker(真正做线程调度的类)

发现有scheduler.scheduleDirect(new Runnable()）点击进去，最终调用
public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
final Worker w = createWorker();

    final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

    w.schedule(new Runnable() {
        @Override
        public void run() {
            try {
                decoratedRun.run();
            } finally {
                w.dispose();
            }
        }
    }, delay, unit);

    return w;
}
最终还是w.schedule(new Runnable()),w就是NewThreadWorker，找到这个类看一下schedule方法,最终会调用：


public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, TimeUnit unit, DisposableContainer parent) {
    Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

    ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);

    if (parent != null) {
        if (!parent.add(sr)) {
            return sr;
        }
    }

    Future<?> f;
    try {
        if (delayTime <= 0) {
            f = executor.submit((Callable<Object>)sr);
        } else {
            f = executor.schedule((Callable<Object>)sr, delayTime, unit);
        }
        sr.setFuture(f);
    } catch (RejectedExecutionException ex) {
        parent.remove(sr);
        RxJavaPlugins.onError(ex);
    }

    return sr;
}

看到future和executor了，这里就是线程切换

observeOn(AndroidSchedulers.mainThread())

ObservableObserveOn 最终调用的是HandlerScheduler和HandlerWorker

HandlerWorker:

@Override
    public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
        if (run == null) throw new NullPointerException("run == null");
        if (delay < 0) throw new IllegalArgumentException("delay < 0: " + delay);
        if (unit == null) throw new NullPointerException("unit == null");

        if (disposed) {
            return Disposables.disposed();
        }

        run = RxJavaPlugins.onSchedule(run);

        ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);

        Message message = Message.obtain(handler, scheduled);
        message.obj = this; // Used as token for batch disposal of this worker's runnables.

        handler.sendMessageDelayed(message, unit.toMillis(delay));

        // Re-check disposed state for removing in case we were racing a call to dispose().
        if (disposed) {
            handler.removeCallbacks(scheduled);
            return Disposables.disposed();
        }

        return scheduled;
    }

发现是用Handler来做线程切换，Handler管理的Looper是Looper.getMainLooper()，所以把消息发送到了主线程。