Rxjava源码解析之监听回调

约定不等于承诺〃 2022-05-27 09:22 209阅读 0赞

# Rxjava源码解析之监听回调 #

--------------------

# 前言 #

> 该篇分析Rxjava接口回调流程的逻辑， 为了方便理解，用最简单的实例分析。

1.  实例代码；
2.  接口回调流程分析；

# 实例代码 #

> 导入第三方Rxjava RxAndroid(用到了UI线程)

compile 'io.reactivex.rxjava2:rxjava:2.0.7'
    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

> 简单的实例如下：

Observable.just("test")
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Observer<String>() {
                        @Override
                        public void onSubscribe(Disposable d) {
    
                        }
                        @Override
                        public void onNext(String o) {
                            // println o: test
                        }
                        @Override
                        public void onError(Throwable e) {
    
                        }
                        @Override
                        public void onComplete() {
    
                        }
                    });

# 接口回调分析 #

> 根据实例代码，我们一行一行分析。

## ObserverableJust ##

1.由实例代码1行：通过传入一个实例调用Observable的静态just(T item）方法生成ObservableJust(item)实例(该实例也是继承Observable);

public static <T> Observable<T> just(T item) {
        ObjectHelper.requireNonNull(item, "The item is null");
        return RxJavaPlugins.onAssembly(new ObservableJust<T>(item));
    }

## ObservableSubscribeOn ##

2.由实例代码2行：由上，observableJust实例的subscribeOn(Scheduler)如下（实际上该子类没有重写该方法，该方法还是父类Observale实现的），传入ObservableSource和scheduler生成ObservableSubscribeOn对象并返回；该方法的ObservableSource是上面的ObservableJust(最终继承ObservableSource)对象, scheduler是创建的IoScheduler对象；

public final Observable<T> subscribeOn(Scheduler scheduler) {
            ObjectHelper.requireNonNull(scheduler, "scheduler is null");
            return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
        }
    
    public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }

## ObservableObserveOn ##

3.由实例代码3行：由上，observableSubscribeOn实例的observeOn(Scheduler)如下（实际上该子类没有重写该方法，该方法还是父类Observale实现的），该方法调用了重载方法observeOn(Scheduler scheduler, boolean delayError, int bufferSize) ；由8行，传入ObservableSource和scheduler生成ObservableObserveOn对象(其他2个参数先不管)并返回；该方法的ObservableSource是上面的ObservableSubscribeOn(最终继承ObservableSource)对象, scheduler是创建的HandlerScheduler对象；

public final Observable<T> observeOn(Scheduler scheduler) {
        return observeOn(scheduler, false, bufferSize());
    }
    
     public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
            ObjectHelper.requireNonNull(scheduler, "scheduler is null");
            ObjectHelper.verifyPositive(bufferSize, "bufferSize");
            return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
        }

4.由实例代码4行：由4行，通过传入this(当前是上一步产生的ObservableObserveOn), observer（我们实例代码创建的)调用RxJavaPlugins的静态onSubscribe(observable, observer)由21-29行，该方法最终返回实例代码传入的observe对象；由6行，调用subscribeActual(observer);该方法ObservableObserveOn实例实现了如33-40行，由39行，通过传入的observer(实例创建的Observer对象）创建一个ObservableObserveOn的内部ObserveOnObserver实例，再通过source.subscribe(observer)把刚刚创建的ObserveOnObserver实例传入；（source是创建ObservableObserveOn传入的observableSubscribeOn对象）

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;
            }
        }
    
      public static <T> Observer<? super T> onSubscribe(@NonNull Observable<T> source, 
      @NonNull Observer<? super T> observer) {
        BiFunction<? super Observable, ? super Observer, ? extends Observer> f = 
        onObservableSubscribe;
        if (f != null) {
            return apply(f, source, observer);
        }
        return observer;
    }
    
    **// ObservableObserveOn实例实现了父类Observable抽象subscribeActual(Observer<? super T> observer) 方法：**
     @Override
        protected void subscribeActual(Observer<? super T> observer) {
            if (scheduler instanceof TrampolineScheduler) {
                source.subscribe(observer);
            } else {
                Scheduler.Worker w = scheduler.createWorker();
    
                source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
            }
        }

5.由上，observableSubscribeOn.subscribe(Observer

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

6.我们看最后一步，由上知道，source.subscribe(observer)方法，最终会通过传入observer调用子类实现的subscribeActual(observer);ObservableJust类的subscribeActual方法如下，首先通过传入observer和value(实例中传入的”test”对象）创建ScalarDisposable, 然后执行run()方法；由9行，调用observer.onNext(value)方法；也是这个最终回调；

protected void subscribeActual(Observer<? super T> s) {
            ScalarDisposable<T> sd = new ScalarDisposable<T>(s, value);
            s.onSubscribe(sd);
            sd.run();
        }
    
         public void run() {
                if (get() == START && compareAndSet(START, ON_NEXT)) {
                    observer.onNext(value);  // 结果回调
                    if (get() == ON_NEXT) {
                        lazySet(ON_COMPLETE);
                        observer.onComplete(); // 回调
                    }
                }
            }

# 最后用一个简陋的流程图表示一下 #

![这里写图片描述][70]

[70]: /images/20220527/0d9c4ab6b26b468bb0c820c06a8c9c06.png