引言

简单阐述RxJava流程源码,RxJava有以下三种流程,向下递增。

• Observable->Observer
• Observable->Operator->Observer
• Observable->Operator->Scheduler->Observer

简单流程

先看一下RxJava简单的代码:

Observable.create(new Observable.OnSubscribe<Boolean>() {
        @Override
        public void call(Subscriber<? super Boolean> s) {
            s.onNext(true);
            s.onCompleted();
        }
    }).subscribe(new Observer<Boolean>() {

                @Override
                public void onCompleted() {
                    _log("On complete");
                    _progress.setVisibility(View.INVISIBLE);
                }

                @Override
                public void onError(Throwable e) {
                    Timber.e(e, "Error in RxJava Demo concurrency");
                    _log(String.format("Boo! Error %s", e.getMessage()));
                    _progress.setVisibility(View.INVISIBLE);
                }

                @Override
                public void onNext(Boolean bool) {
                    _log(String.format("onNext with return value \"%b\"", bool));
                }
            });

据上:Obervable create传入Observable.OnSubscribe作为参数。OnSubscribe 会被存储在返回的 Observable 对象中，它的作用相当于一个计划表，当Observable被订阅的时候，即当Obervable.subscribe(observer)的时候,OnSubscribe 的 call() 方法会自动被调用，事件序列就会依照设定依次触发。

看下subscribe method 源码

rxjava_subscribe.jpg

将源码简化一下就是这样:

public Subscription subscribe(Subscriber subscriber) {
  subscriber.onStart();//可选的准备方法
  onSubscribe.call(subscriber);//开始执行计划表
  return subscriber;//返回Subscription 方便unsubscribe()
}

你会发现触发Observable的时间是subscribe产生订阅关系的时候。
另外你会看到subscribe中参数是Subscriber类型,Subscriber是实现Oberver的抽象类。

带有Operator的流程

这里我们说的Operator是一种抽象的概念,上述简单流程中我们是手写计划表OnSubscribe,而这里我们不需要,我们只需要调用just,map,等转换操作,这些操作内部帮我们实现了OnSubscribe计划表,我将这些操作称之为Operator.

我们看一下带有Operator的简单代码:

Observable.just(true).map(new Func1<Boolean, Boolean>() {
        @Override
        public Boolean call(Boolean aBoolean) {
            _log("Within Observable");
            _doSomeLongOperation_thatBlocksCurrentThread();
            return aBoolean;
        }
    }).subscribe(new Observer<Boolean>() {

        @Override
        public void onCompleted() {
            _log("On complete");
            _progress.setVisibility(View.INVISIBLE);
        }

        @Override
        public void onError(Throwable e) {
            Timber.e(e, "Error in RxJava Demo concurrency");
            _log(String.format("Boo! Error %s", e.getMessage()));
            _progress.setVisibility(View.INVISIBLE);
        }

        @Override
        public void onNext(Boolean bool) {
            _log(String.format("onNext with return value \"%b\"", bool));
        }
    });

据上:和第一种简单流程的代码相比,subscribe之后的代码是一致的,不同的是subscribe之前的代码。两个方法just和map method,通过源码分析下功能。

• just method

rxjava_just_method.jpg rxjava_ScalarSynchronousObservable.jpg

just function:将传入的参数发送给Subscriber订阅者。
你可以根据上面ScalarSynchronousObservable的构造函数,重写OnSubscribe计划表看出来。

• map method

rxjava_map.jpg
rxjava_lift_method.jpg
rxjava_OperatorMap.jpg

将上面代码简化就是:

public <R> Observable<R> lift(Operator<? extends R, ? super T> operator) {
return Observable.create(new OnSubscribe<R>() {
    @Override
    public void call(Subscriber subscriber) {
        Subscriber newSubscriber = operator.call(subscriber);//通过Operator创建一个新的Subscriber,通过OperatorMap将新旧Subscriber建立一种关系。
        newSubscriber.onStart();//准备操作
        onSubscribe.call(newSubscriber);//onSubscribe计划表唤醒新的Subscriber,新的Subscriber会联系subscribe订阅的Subscriber(订阅者)。
    }
});
}   

借助抛物线文章中的图抽象表现出来:

rxjava_map_pic.jpg

再带上Scheduler的流程

Scheduler function:指明Observable和Observer是运行在哪个线程中。

• subscribeOn指定的是Observable(被观察者)所在线程
• observeOn指定的是Observer(观察者)所在线程

我们在带有Operator的基础上加上Scheduler.

Observable.just(true).map(new Func1<Boolean, Boolean>() {
        @Override
        public Boolean call(Boolean aBoolean) {
            _log("Within Observable");
            _doSomeLongOperation_thatBlocksCurrentThread();
            return aBoolean;
        }
    }).subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread()).subscribe(...)

observeOn与subscribeOn也是通过lift变换原理实现线程切换, observeOn切换线程是发生在lift方式中的内建的Subscriber中,subscribeOn切换线程是发生在OnSubscribe计划表中。

调度器源码我也没太搞明白,以后有机会补上。不过调度器作为RxJava的一大特性,在Android编程中使用非常方便。

总结

• RxJava最核心流程是Observable->Oberver 当subscribe订阅的时候,不管增加怎样复杂的变化,一定是OnSubscribe计划表来通知观察者,被观察者发生变化。
• RxJava灵活的特性主要在两个方面,数据序列的变化机制和线程切换。

参考

• 抛物线的一篇文章