RXJava2.0源码浅析

作者: 向洋xyang | 来源:发表于2018-07-28 11:27 被阅读177次

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

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)

第一个方法中回调回来的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

第二个方法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>() {...})

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

结束

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

RXJava2.0源码浅析

RXJava是什么

为什么使用RXJava

观察者模式

RXJava基本使用

总结

结束

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Android项目源码分析

RxJava