RXJava2.0源码浅析

image.png

RXJava是什么

RxJava 在 GitHub 主页上官方解释为：RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）

为什么使用RXJava

上面说了RXJava是一个异步事件库，Android中关于异步的处理有Handler消息机制、AsyncTask异步任务，但是使用RXJava来实行异步处理更加简洁，链式调用书写起来更方便，逻辑也更加清晰，而且RXJava有很多适合各种场景的操作符，功能非常强大，使用起来非常方便。

观察者模式

观察者模式面向的需求是：A 对象（观察者）对 B 对象（被观察者）的某种变化高度敏感，A对象需要在 B 变化的一瞬间做出反应。
RxJava 有两个基本概念：Observable (可观察者，即被观察者)、 Observer (观察者)、 subscribe (订阅)。Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知 Observer。

RXJava基本使用

Observable.create(new ObservableOnSubscribe<TestBean>() {
            @Override
            public void subscribe(ObservableEmitter<TestBean> e) throws Exception {
                Log.d(TAG, "---subscribe---");
                e.onNext(new TestBean());
                e.onComplete();
            }
        })
                .map(new Function<TestBean, String>() {
                    @Override
                    public String apply(TestBean testBean) throws Exception {
                        Log.d(TAG, "---map操作符---");
                        return "test";
                    }
                })
//                .subscribeOn(Schedulers.io())
//                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        Log.d(TAG, "---onSubscribe---");
                    }

                    @Override
                    public void onNext(@NonNull String string) {
                        Log.d(TAG, "---onNext---");
                    }

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

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "---onComplete---");
                    }
                });
    }
  1. Observable.create(...)产生一个继承抽象类Observable的子类对象（ObservableCreate对象），RxJavaPlugins.onAssembly(...)为hook相关操作，一般的调用f==null，直接返回传进去的参数，即ObservableCreate对象
public abstract class Observable<T> implements ObservableSource<T> {

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

   @SuppressWarnings({ "rawtypes", "unchecked" })
   @NonNull
   public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
       Function<? super Observable, ? extends Observable> f = onObservableAssembly;
       if (f != null) {
           return apply(f, source);
       }
       return source;
   }

2.Observable.map(...)方法会产生一个ObservableMap对象，此对象的构造方法会保存this(这里的this为Observable对象（多态），即上面返回的ObservableCreate对象)和传进来的mapper。

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

3.Observable.subscribe(...)让观察者和被观察者产生订阅关系（第1次订阅)，在里面会调用subscribeActual(observer)方法,并将第1个观察者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;
       }
   }

subscribeActual(observer)为Observable的抽象方法，实际上会执行子类ObservableMap的subscribeActual(observer)如下:

 @Override
    public void subscribeActual(Observer<? super U> t) {
        source.subscribe(new MapObserver<T, U>(t, function));
    }
  • 该方法中的source为ObservableMap构造函数中保存的Observable的子类（多态），即ObservableCreate对象
  • source.subscribe(new MapObserver<T, U>(t, function))，又产生了一次订阅（第2次订阅)，和上面的第3步一样，会走Observable的抽象方法subscribeActual(observer)，并将MapObserver对象（第2个观察者）回调回去，其中MapObserver对象的构造方法中会保存t（第1次订阅回调过来的的observer）和function(mapper)
  • 实际会执行ObservableCreate对象中的subscribeActual(observer)方法如下：
@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);
        }
    }

该方法会执行两个方法observer.onSubscribe(parent),source.subscribe(parent)

  1. 第一个方法中回调回来的observer为上面的mapObserver,
    parent为上面的CreateEmitter对象（下面再介绍CreateEmitter里面做了什么）
    Observer是一个接口,调用onSubscribe(parent)实际走的是mapObserver的父类BasicFuseableObserver（Observer的实现类）中的onSubscribe(...)方法,如下：
public abstract class BasicFuseableObserver<T, R> implements Observer<T>, QueueDisposable<R> {

            ......

    @SuppressWarnings("unchecked")
    @Override
    public final void onSubscribe(Disposable s) {
        if (DisposableHelper.validate(this.s, s)) {
            this.s = s;
            if (s instanceof QueueDisposable) {
                this.qs = (QueueDisposable<T>)s;
            }
            if (beforeDownstream()) {
                actual.onSubscribe(this);
                afterDownstream();
            }
        }
    }
              ......

}

其中actual为在MapObserver构造方法中保存的observer即第一次回调回来的observer，并将mapObserver回调回去
终于回到了最初第一次new的observer，接着就走到了下图中的onSubscribe(@NonNull Disposable d),d为回调回来的mapObserver对象。

image1.png

  1. 第二个方法source.subscribe(parent),source为上图中new出来的ObservableOnSubscribe匿名对象，parent为第一个方法里的CreateEmitter对象，执行回调方法subscribe(ObservableEmitter<TestBean> e)，回调回来的ObservableEmitter对象（CreateEmitter对象）,CreateEmitter构造中保存了第2次订阅回调回来的观察者MapObserver对象如下图， CreateEmitter对象可以调用我们常用的方法如事件的发射emitter.onNext(...)和完成事件emitter.onComplete()
  • 首先看emitter.onNext(...)做了什么操作

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为第2次订阅回调回来的观察者mapObserver，调用他的onNext(T t)方法，在onNext(T t)方法中会执行mapper.apply(t)并返回一个泛型U v,实际会回调到下图中箭头所指的apply(...)方法

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);
        }
}
image2.png

紧接着又会调用actual.onNext(v),最终会回到第一次订阅回调回来的观察者的onNext()。

  • 接着当调用emitter.onComplete()时会走到mapObserver父类BasicFuseableObserver的onComplete()，其中actual为第一次回调回来的观察者observer，最终会回到它的onComplete() 
@Override
    public void onComplete() {
        if (done) {
            return;
        }
        done = true;
        actual.onComplete();
    }

到此为止一个简单的调用流程就走完了，当被观察者和观察者产生订阅关系时，各个方法的调用顺序为下图中的1、2、3、4、5、6

image3.png

总结

Observable.create(new ObservableOnSubscribe<TestBean>() {...})
                .map(new Function<TestBean, String>(){...} )
                //.subscribeOn(Schedulers.io())
               // .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {...})
  1. Observable通过.XXX(...)都会依次生成一个Observable对象（多态），每个Observable对象都会依次保存上一个Observable对象的引用和传进来的参数，参数最终会被保存到观察者observer中（链式调用连起来的关键）。
  2. 当通过Observable.subscribe(new Observer<String>(){...})产生订阅时，抽象类Observable会在subscribe(observer)方法中调用subscribeActual(observer)(子类已复写该方法),实际上就是Observable子类(ObservableCreate、ObservableMap等均继承Observable)执行此方法，在subscribeActual(observer)方法中会调用source.subscribe(parent)利用保存的上个Observable(source)又一次产生订阅，也就是自下而上产生订阅直到第一个source开始发射事件。
  3. subscribeActual(observer)会生成一个新的observer(如MapObserver对象),新的observer会自下而上的保存上个observer,当发射事件时会利用observer 自上而下将事件依次发送到最下面的observer上。

结束

第一次写blog,表达能力也不好，希望用来记录自己的学习过程，以上的RXJava只是最基本的调用（注释了线程切换，如果加上线程切换整体流程大同小异），有不对的地方欢迎指正，下次准备学习RXJava的Schedulers线程调度。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 212,718评论 6赞 492
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,683评论 3赞 385
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 158,207评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,755评论 1赞 284
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,862评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,050评论 1赞 291
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,136评论 3赞 410
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,882评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,330评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,651评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,789评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,477评论 4赞 333
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,135评论 3赞 317
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,864评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,099评论 1赞 267
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,598评论 2赞 362
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,697评论 2赞 351