从微观角度解读 RxJava 源码 (RxJava 系列 6)

1,161 阅读13分钟

转载请注明出处:www.jianshu.com/p/bf56e0349…

前言

通过前面五个篇幅的介绍,相信大家对RxJava的基本使用以及操作符应该有了一定的认识。但是知其然还要知其所以然;所以从这一章开始我们聊聊源码,分析RxJava的实现原理。本文我们主要从三个方面来分析RxJava的实现:

  • RxJava基本流程分析
  • 操作符原理分析
  • 线程调度原理分析

本章节基于RxJava1.1.9版本的源码

一、RxJava执行流程分析

RxJava系列2(基本概念及使用介绍)中我们介绍过,一个最基本的RxJava调用是这样的:

示例A

Observable.create(new Observable.OnSubscribe() {
    @Override
    public void call(Subscriber subscriber) {
        subscriber.onNext("Hello RxJava!");
        subscriber.onCompleted();
    }
}).subscribe(new Subscriber() {
    @Override
    public void onCompleted() {
        System.out.println("completed!");
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onNext(String s) {
        System.out.println(s);
    }
});

首先调用Observable.create()创建一个被观察者Observable,同时创建一个OnSubscribe作为create()方法的入参;接着创建一个观察者Subscriber,然后通过subseribe()实现二者的订阅关系。这里涉及到三个关键对象和一个核心的方法:

  • Observable(被观察者)
  • OnSubscribe (从纯设计模式的角度来理解,OnSubscribe.call()可以看做是观察者模式中被观察者用来通知观察者的notifyObservers()方法)
  • Subscriber (观察者)
  • subscribe() (实现观察者与被观察者订阅关系的方法)

1、Observable.create()源码分析

首先我们来看看Observable.create()的实现:

public static  Observable create(OnSubscribe f) {
    return new Observable(RxJavaHooks.onCreate(f));
}

这里创建了一个被观察者Observable,同时将RxJavaHooks.onCreate(f)作为构造函数的参数,源码如下:

protected Observable(OnSubscribe f) {
    this.onSubscribe = f;
}

我们看到源码中直接将参数RxJavaHooks.onCreate(f)赋值给了当前我们构造的被观察者Observable的成员变量onSubscribe。那么RxJavaHooks.onCreate(f)返回的又是什么呢?我们接着往下看:

public static  Observable.OnSubscribe onCreate(Observable.OnSubscribe onSubscribe) {
    Func1 f = onObservableCreate;
    if (f != null) {
        return f.call(onSubscribe);
    }
    return onSubscribe;
}

由于我们并没调用RxJavaHooks.initCreate(),所以上面代码中的onObservableCreate为null;因此RxJavaHooks.onCreate(f)最终返回的就是f,也就是我们在Observable.create()的时候new出来的OnSubscribe。(由于对RxJavaHooks的理解并不影响我们对RxJava执行流程的分析,因此在这里我们不做进一步的探讨。为了方便理解我们只需要知道RxJavaHooks一系列方法的返回值就是入参本身就OK了,例如这里的RxJavaHooks.onCreate(f)返回的就是f)。

至此我们做下逻辑梳理:Observable.create()方法构造了一个被观察者Observable对象,同时将new出来的OnSubscribe赋值给了该Observable的成员变量onSubscribe

2、Subscriber源码分析

接着我们看下观察者Subscriber的源码,为了增加可读性,我去掉了源码中的注释和部分代码。

public abstract class Subscriber implements Observer, Subscription {

    private final SubscriptionList subscriptions;//订阅事件集,所有发送给当前Subscriber的事件都会保存在这里

    ...

    protected Subscriber(Subscriber subscriber, boolean shareSubscriptions) {
        this.subscriber = subscriber;
        this.subscriptions = shareSubscriptions && subscriber != null ? subscriber.subscriptions : new SubscriptionList();
    }

    ...

    @Override
    public final void unsubscribe() {
        subscriptions.unsubscribe();
    }

    @Override
    public final boolean isUnsubscribed() {
        return subscriptions.isUnsubscribed();
    }

    public void onStart() {
    }

    ...
}
public interface Subscription {
    void unsubscribe();
    boolean isUnsubscribed();
}

Subscriber实现了Subscription接口,从而对外提供isUnsubscribed()unsubscribe()方法。前者用于判断是否已经取消订阅;后者用于将订阅事件列表(也就是当前观察者的成员变量subscriptions)中的所有Subscription取消订阅,并且不再接受观察者Observable发送的后续事件。

3、subscribe()源码分析

前面我们分析了观察者和被观察者相关的源码,那么接下来便是整个订阅流程中最最关键的环节了。

public final Subscription subscribe(Subscriber subscriber) {
    return Observable.subscribe(subscriber, this);
}
static  Subscription subscribe(Subscriber subscriber, Observable observable) {

    ...

    subscriber.onStart();

    if (!(subscriber instanceof SafeSubscriber)) {
        subscriber = new SafeSubscriber(subscriber);
    }

    try {
        RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);

        return RxJavaHooks.onObservableReturn(subscriber);
    } catch (Throwable e) {
        ...
        return Subscriptions.unsubscribed();
    }
}

subscribe()方法中将传进来的subscriber包装成了SafeSubscriberSafeSubscriber其实是subscriber的一个代理,对subscriber的一系列方法做了更加严格的安全校验。保证了onCompleted()onError()只会有一个被执行且只执行一次,一旦它们其中方法被执行过后onNext()就不在执行了。

上述代码中最关键的就是RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber)。这里的RxJavaHooks和之前提到的一样,RxJavaHooks.onObservableStart(observable, observable.onSubscribe)返回的正是他的第二个入参observable.onSubscribe,也就是当前observable的成员变量onSubscribe。而这个成员变量我们前面提到过,它是我们在Observable.create()的时候new出来的。所以这段代码可以简化为onSubscribe.call(subscriber)。这也印证了我在RxJava系列2(基本概念及使用介绍)中说的,onSubscribe.call(subscriber)中的subscriber正是我们在subscribe()方法中new出来的观察者。

到这里,我们对RxJava的执行流程做个总结:首先我们调用crate()创建一个观察者,同时创建一个OnSubscribe作为该方法的入参;接着调用subscribe()来订阅我们自己创建的观察者Subscriber
一旦调用subscribe()方法后就会触发执行OnSubscribe.call()。然后我们就可以在call方法调用观察者subscriberonNext(),onCompleted(),onError()

最后我用张图来总结下之前的分析结果:


RxJava基本流程分析

二、操作符原理分析

之前我们介绍过几十个操作符,要一一分析它们的源码显然不太现实。在这里我抛砖引玉,选取一个相对简单且常用的map操作符来分析。

我们先来看一个map操作符的简单应用:

示例B

Observable.create(new Observable.OnSubscribe() {
    @Override
    public void call(Subscriber subscriber) {
        subscriber.onNext(1);
        subscriber.onCompleted();
    }
}).map(new Func1() {
    @Override
    public String call(Integer integer) {
        return "This is " + integer;
    }
}).subscribe(new Subscriber() {
    @Override
    public void onCompleted() {
        System.out.println("onCompleted!");
    }
    @Override
    public void onError(Throwable e) {
        System.out.println(e.getMessage());
    }
    @Override
    public void onNext(String s) {
        System.out.println(s);
    }
});

为了便于表述,我将上面的代码做了如下拆解:

Observable observableA = Observable.create(new Observable.OnSubscribe() {
    @Override
    public void call(Subscriber subscriber) {
        subscriber.onNext(1);
        subscriber.onCompleted();
    }
});

Subscriber subscriberOne = new Subscriber() {
    @Override
    public void onCompleted() {
        System.out.println("onCompleted!");
    }
    @Override
    public void onError(Throwable e) {
        System.out.println(e.getMessage());
    }
    @Override
    public void onNext(String s) {
        System.out.println(s);
    }
};

Observable observableB = 
        observableA.map(new Func1() {
                @Override
                public String call(Integer integer) {
                    return "This is " + integer;;
                }
            });

observableB.subscribe(subscriberOne);

map()的源码和上一小节介绍的create()一样位于Observable这个类中。

public final  Observable map(Func1 func) {
    return create(new OnSubscribeMap(this, func));
}

通过查看源码我们发现调用map()的时候实际上是创建了一个新的被观察者Observable,我们姑且称它为ObservableB;一开始通过Observable.create()创建的Observable我们称之为ObservableA。在创建ObservableB的时候同时创建了一个OnSubscribeMap,而ObservableA和变换函数Func1则作为构造OnSubscribeMap的参数。

public final class OnSubscribeMap implements OnSubscribe {

    final Observable source;//ObservableA

    final Func1 transformer;//map操作符中的转换函数Func1。T为转换前的数据类型,在上面的例子中为Integer;R为转换后的数据类型,在该例中为String。

    public OnSubscribeMap(Observable source, Func1 transformer) {
        this.source = source;
        this.transformer = transformer;
    }

    @Override
    public void call(final Subscriber o) {//结合第一小节的分析结果,我们知道这里的入参o其实就是我们自己new的观察者subscriberOne。
        MapSubscriber parent = new MapSubscriber(o, transformer);
        o.add(parent);
        source.unsafeSubscribe(parent);
    }

    static final class MapSubscriber extends Subscriber {

        final Subscriber actual;//这里的actual就是我们在调用subscribe()时创建的观察者mSubscriber
        final Func1 mapper;//变换函数
        boolean done;

        public MapSubscriber(Subscriber actual, Func1 mapper) {
            this.actual = actual;
            this.mapper = mapper;
        }

        @Override
        public void onNext(T t) {
            R result;
            try {
                result = mapper.call(t);
            } catch (Throwable ex) {
                Exceptions.throwIfFatal(ex);
                unsubscribe();
                onError(OnErrorThrowable.addValueAsLastCause(ex, t));
                return;
            }
            actual.onNext(result);
        }

        @Override
        public void onError(Throwable e) {
            ...
            actual.onError(e);
        }

        @Override
        public void onCompleted() {
            ...
            actual.onCompleted();
        }

        @Override
        public void setProducer(Producer p) {
            actual.setProducer(p);
        }
    }
}

OnSubscribeMap实现了OnSubscribe接口,因此OnSubscribeMap就是一个OnSubscribe。在调用map()的时候创建了一个新的被观察者ObservableB,然后我们用ObservableB.subscribe(subscriberOne)订阅了观察者subscriberOne。结合我们在第一小节的分析结果,所以OnSubscribeMap.call(o)中的o就是subscribe(subscriberOne)中的subscriberOne;一旦调用了ObservableB.subscribe(subscriberOne)就会执行OnSubscribeMap.call()

call()方法中,首先通过我们的观察者o和转换函数transformer构造了一个MapSubscriber,最后调用了source也就是observableAunsafeSubscribe()方法。即observableA订阅了一个观察者MapSubscriber

public final Subscription unsafeSubscribe(Subscriber subscriber) {
    try {
        ...
        RxJavaHooks.onObservableStart(this, onSubscribe).call(subscriber);
        return RxJavaHooks.onObservableReturn(subscriber);
    } catch (Throwable e) {
        ...
        return Subscriptions.unsubscribed();
    }
}

上面这段代码最终执行了onSubscribe也就是OnSubscribeMapcall()方法,call()方法中的参数就是之前在OnSubscribeMap.call()中new出来的MapSubscriber。最后在call()方法中执行了我们自己的业务代码:

subscriber.onNext(1);
subscriber.onCompleted();

其实也就是执行了MapSubscriberonNext()onCompleted()

@Override
public void onNext(T t) {
    R result;
    try {
        result = mapper.call(t);
    } catch (Throwable ex) {
        ...
        return;
    }
    actual.onNext(result);
}

onNext(T t)方法中的的mapper就是变换函数,actual就是我们在调用subscribe()时创建的观察者subscriberOne。这个T就是我们例子中的IntegerR就是String。在onNext()中首先调用变换函数mapper.call()T转换成R(在我们的例子中就是将Integer类型的1转换成了String类型的“This is 1”);接着调用subscriberOne.onNext(String result)。同样在调用MapSubscriber.onCompleted()时会执行subscriberOne.onCompleted()。这样就完成了一直完成的调用流程。

我承认太啰嗦了,花费了这么大的篇幅才将map()的转换原理解释清楚。我也是希望尽量的将每个细节都呈现出来方便大家理解,如果看我啰嗦了这么久还是没能理解,请看下面我画的这张执行流程图。


加入Map操作符后的执行流程

三、线程调度原理分析

在前面的文章中我介绍过RxJava可以很方便的通过subscribeOn()observeOn()来指定数据流的每一部分运行在哪个线程。其中subscribeOn()指定了处理Observable的全部的过程(包括发射数据和通知)的线程;observeOn()指定了观察者的onNext(), onError()onCompleted()执行的线程。接下来我们就分析分析源码,看看线程调度是如何实现的。

在分析源码前我们先看看一段常见的通过RxJava实现的线程调度代码:

示例C

Observable.create(new Observable.OnSubscribe() {
    @Override
    public void call(Subscriber subscriber) {
        subscriber.onNext("Hello RxJava!");
        subscriber.onCompleted();
    }
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber() {
    @Override
    public void onCompleted() {
        System.out.println("completed!");
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onNext(String s) {
        System.out.println(s);
    }
});

1、subscribeOn()源码分析

public final Observable subscribeOn(Scheduler scheduler) {
    ...
    return create(new OperatorSubscribeOn(this, scheduler));
}

通过上面的代码我们可以看到,subscribeOn()map()一样是创建了一个新的被观察者Observable。因此我大致就能猜到subscribeOn()的执行流程应该和map()差不多,OperatorSubscribeOn肯定也是一个OnSubscribe。那我们接下来就看看OperatorSubscribeOn的源码:

public final class OperatorSubscribeOn implements OnSubscribe {

    final Scheduler scheduler;//线程调度器,用来指定订阅事件发送、处理等所在的线程
    final Observable source;

    public OperatorSubscribeOn(Observable source, Scheduler scheduler) {
        this.scheduler = scheduler;
        this.source = source;
    }

    @Override
    public void call(final Subscriber subscriber) {
        final Worker inner = scheduler.createWorker();
        subscriber.add(inner);

        inner.schedule(new Action0() {
            @Override
            public void call() {
                final Thread t = Thread.currentThread();

                Subscriber s = new Subscriber(subscriber) {
                    @Override
                    public void onNext(T t) {
                        subscriber.onNext(t);
                    }

                    @Override
                    public void onError(Throwable e) {
                        try {
                            subscriber.onError(e);
                        } finally {
                            inner.unsubscribe();
                        }
                    }

                    @Override
                    public void onCompleted() {
                        try {
                            subscriber.onCompleted();
                        } finally {
                            inner.unsubscribe();
                        }
                    }

                    @Override
                    public void setProducer(final Producer p) {
                        subscriber.setProducer(new Producer() {
                            @Override
                            public void request(final long n) {
                                if (t == Thread.currentThread()) {
                                    p.request(n);
                                } else {
                                    inner.schedule(new Action0() {
                                        @Override
                                        public void call() {
                                            p.request(n);
                                        }
                                    });
                                }
                            }
                        });
                    }
                };
                source.unsafeSubscribe(s);
            }
        });
    }
}

OperatorSubscribeOn实现了OnSubscribe接口,call()中对Subscriber的处理也和OperatorMapSubscriber的处理类似。首先通过scheduler构建了一个Worker;然后用传进来的subscriber构造了一个新的Subscriber s,并将s丢到Worker.schedule()中来处理;最后用原Observable去订阅观察者s。而这个Worker就是线程调度的关键!前面的例子中我们通过subscribeOn(Schedulers.io())指定了Observable发射处理事件以及通知观察者的一系列操作的执行线程,正是通过这个Schedulers.io()创建了我们前面提到的Worker。所以我们来看看Schedulers.io()的实现。

首先通过Schedulers.io()获得了ioScheduler并返回,上面的OperatorSubscribeOn通过这个的SchedulercreateWorker()方法创建了我们前面提到的Worker

public static Scheduler io() {
    return RxJavaHooks.onIOScheduler(getInstance().ioScheduler);
}

接着我们看看这个ioScheduler是怎么来的,下面的代码向我们展现了是如何在Schedulers的构造函数中通过RxJavaSchedulersHook.createIoScheduler()来初始化ioScheduler的。

private Schedulers() {

    ...

    Scheduler io = hook.getIOScheduler();
    if (io != null) {
        ioScheduler = io;
    } else {
        ioScheduler = RxJavaSchedulersHook.createIoScheduler();
    }

    ...
}

最终RxJavaSchedulersHook.createIoScheduler()返回了一个CachedThreadScheduler,并赋值给了ioScheduler

public static Scheduler createIoScheduler() {
    return createIoScheduler(new RxThreadFactory("RxIoScheduler-"));
}
public static Scheduler createIoScheduler(ThreadFactory threadFactory) {
    ...
    return new CachedThreadScheduler(threadFactory);
}

到这一步既然我们知道了ioScheduler就是一个CachedThreadScheduler,那我们就来看看它的createWorker()的实现。

public Worker createWorker() {
    return new EventLoopWorker(pool.get());
}

上面的代码向我们赤裸裸的呈现了前面OperatorSubscribeOn中的Worker其实就是EventLoopWorker。我们重点要关注的是他的scheduleActual()

static final class EventLoopWorker extends Scheduler.Worker implements Action0 {
    private final CompositeSubscription innerSubscription = new CompositeSubscription();
    private final CachedWorkerPool pool;
    private final ThreadWorker threadWorker;
    final AtomicBoolean once;

    EventLoopWorker(CachedWorkerPool pool) {
        this.pool = pool;
        this.once = new AtomicBoolean();
        this.threadWorker = pool.get();
    }

    ...

    @Override
    public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
        ...
        ScheduledAction s = threadWorker.scheduleActual(new Action0() {
            @Override
            public void call() {
                if (isUnsubscribed()) {
                    return;
                }
                action.call();
            }
        }, delayTime, unit);
        innerSubscription.add(s);
        s.addParent(innerSubscription);
        return s;
    }
}

通过对源码的一步步追踪,我们知道了前面OperatorSubscribeOn.call()中的inner.schedule()最终会执行到ThreadWorkerscheduleActual()方法。

public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit) {
    Action0 decoratedAction = RxJavaHooks.onScheduledAction(action);
    ScheduledAction run = new ScheduledAction(decoratedAction);
    Future f;
    if (delayTime <= 0) {
        f = executor.submit(run);
    } else {
        f = executor.schedule(run, delayTime, unit);
    }
    run.add(f);
    return run;
}

scheduleActual()中的ScheduledAction实现了Runnable接口,通过线程池executor最终实现了线程切换。上面便是subscribeOn(Schedulers.io())实现线程切换的全部过程。

2、observeOn()源码分析

observeOn()切换线程是通过lift来实现的,相比subscribeOn()在实现原理上相对复杂些。不过本质上最终还是创建了一个新的Observable

public final Observable observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
    ...
    return lift(new OperatorObserveOn(scheduler, delayError, bufferSize));
}

public final  Observable lift(final Operator operator) {
    return create(new OnSubscribeLift(onSubscribe, operator));
}

OperatorObserveOn作为OnSubscribeLift构造函数的参数用来创建了一个新的OnSubscribeLift对象,接下来我们看看OnSubscribeLift的实现:

public final class OnSubscribeLift implements OnSubscribe {

    final OnSubscribe parent;

    final Operator operator;

    public OnSubscribeLift(OnSubscribe parent, Operator operator) {
        this.parent = parent;
        this.operator = operator;
    }

    @Override
    public void call(Subscriber o) {
        try {
            Subscriber st = RxJavaHooks.onObservableLift(operator).call(o);
            try {
                st.onStart();
                parent.call(st);
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                st.onError(e);
            }
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            o.onError(e);
        }
    }
}

OnSubscribeLift继承自OnSubscribe,通过前面的分析我们知道一旦调用了subscribe()将观察者与被观察绑定后就会触发被观察者所对应的OnSubscribecall()方法,所以这里会触发OnSubscribeLift.call()。在call()中调用了OperatorObserveOn.call()并返回了一个新的观察者Subscriber st,接着调用了前一级Observable对应OnSubscriber.call(st)

我们再看看OperatorObserveOn.call()的实现:

public Subscriber call(Subscriber child) {
    ...
    ObserveOnSubscriber parent = new ObserveOnSubscriber(scheduler, child, delayError, bufferSize);
    parent.init();
    return parent;
}

OperatorObserveOn.call()中创建了一个ObserveOnSubscriber并调用init()进行了初始化。

static final class ObserveOnSubscriber extends Subscriber implements Action0 {

    ...

    @Override
    public void onNext(final T t) {
        ...
        schedule();
    }

    @Override
    public void onCompleted() {
        ...
        schedule();
    }

    @Override
    public void onError(final Throwable e) {
        ...
        schedule();
    }

    protected void schedule() {
        if (counter.getAndIncrement() == 0) {
            recursiveScheduler.schedule(this);
        }
    }

    @Override
    public void call() {
        long missed = 1L;
        long currentEmission = emitted;

        final Queue q = this.queue;
        final Subscriber localChild = this.child;
        final NotificationLite localOn = this.on;

        for (;;) {
            long requestAmount = requested.get();

            while (requestAmount != currentEmission) {
                boolean done = finished;
                Object v = q.poll();
                boolean empty = v == null;

                if (checkTerminated(done, empty, localChild, q)) {
                    return;
                }

                if (empty) {
                    break;
                }

                localChild.onNext(localOn.getValue(v));

                currentEmission++;
                if (currentEmission == limit) {
                    requestAmount = BackpressureUtils.produced(requested, currentEmission);
                    request(currentEmission);
                    currentEmission = 0L;
                }
            }

            if (requestAmount == currentEmission) {
                if (checkTerminated(finished, q.isEmpty(), localChild, q)) {
                    return;
                }
            }

            emitted = currentEmission;
            missed = counter.addAndGet(-missed);
            if (missed == 0L) {
                break;
            }
        }
    }

    ...
}

ObserveOnSubscriber继承自Subscriber,并实现了Action0接口。我们看到ObserveOnSubscriberonNext()onCompleted()onError()都有个schedule(),这个方法就是我们线程调度的关键;通过schedule()将新观察者ObserveOnSubscriber发送给subscriberOne的所有事件都切换到了recursiveScheduler所对应的线程,简单的说就是把subscriberOneonNext()onCompleted()onError()方法丢到了recursiveScheduler对应的线程中来执行。

那么schedule()又是如何做到这一点的呢?他内部调用了recursiveScheduler.schedule(this)recursiveScheduler其实就是一个Worker,和我们在介绍subscribeOn()时提到的worker一样,执行schedule()实际上最终是创建了一个runable,然后把这个runnable丢到了特定的线程池中去执行。在runnablerun()方法中调用了ObserveOnSubscriber.call(),看上面的代码大家就会发现在call()方法中最终调用了subscriberOneonNext()onCompleted()onError()方法。这便是它实现线程切换的原理。

好了,我们最后再看看示例C对应的执行流程图,帮助大家加深理解。


RxJava执行流程

总结

这一章以执行流程操作符实现以及线程调度三个方面为切入点剖析了RxJava源码。下一章将站在更宏观的角度来分析整个RxJava的框架结构、设计思想等等。敬请期待~~ :)

如果大家喜欢这一系列的文章,欢迎关注我的知乎专栏、GitHub、简书博客。