Rxjava 源码解析 - 线程切换源码

Posted 2022-12-02 许佳佳233

Rxjava源码解析系列：
Rxjava 源码解析(一) - subscribe源码
Rxjava 源码解析(二) - 线程切换源码
Rxjava 源码解析(三) - Schedulers默认线程池

概述

前文已经讲了rxjava 简单subscribe的源码，有兴趣的读者可以看下。
Rxjava 源码解析(一) - subscribe源码

本文将基于前文的分析，继续讲下rxjava中断线程切换。

Demo

主要逻辑如下：

1. Observable调用create()创建 ObservableCreate
2. ObservableCreate调用map，返回ObservableMap
3. ObservableMap调用subscribeOn，返回ObservableSubscribeOn
4. 返回ObservableSubscribeOn调用observeOn，返回ObservableSubscribeOn
5. ObservableSubscribeOn调用subscribe，执行最终的逻辑。

Observable
        .create(new ObservableOnSubscribe<String>() 

          @Override
          public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable 
            Log.i("RxJavaTest", "subscribe");
            emitter.onNext("123");
          
        ).map(new Function<String, String>() 

      @Override
      public String apply(String s) throws Throwable 
        return s + "456";
      
    )
        .subscribeOn(Schedulers.io())
        .observeOn(Schedulers.io())
        .subscribe(new Observer<String>() 

          @Override
          public void onSubscribe(@NonNull Disposable d) 
            Log.i("RxJavaTest", "onSubscribe");
          

          @Override
          public void onNext(@NonNull String s) 
            Log.i("RxJavaTest", "onNext: " + s);
          

          @Override
          public void onError(@NonNull Throwable e) 
            Log.i("RxJavaTest", "onError");
          

          @Override
          public void onComplete() 
            Log.i("RxJavaTest", "onComplete");
          
        );

运行结果

RxJavaTest: onSubscribe
RxJavaTest: subscribe
RxJavaTest: onNext: 123456

原理图

上图是根据demo整理的关键逻辑的流程图。
主要内容有以下几点：

• 每个Observable会对应一个Observer，Observer是Observable的内部类，比如MapObserver就是ObservableMap的内部类。
• 当前的Observable会存储上一个Observable，即source对象。比如在Demo
  中ObservableMap的source就是ObservableCreate。
• Observer会存储上一个Observer，即downstream。比如在Demo中，MapObserver的downstream就是CreateEmitter。
• 从调用第一个subscribe方法起，会一层一层的去调用Observable的subscribe方法，最终调用到的是Observable的subscribeActual方法，这个每个子类的逻辑不一样。
  比如ObservableObserveOn的subscribeActual方法就会去获取到执行Observer的Worker然后传给source。
  再比如ObservableMap会将定义map方法的Function对象传给source。
• 此处举例使用的是Observer.onNext方法，当然也可以是onError等其他方法。每个Observer会引用上一个Observer，即downstream。在Demo中最后一个Observer就是ObserveOnObserver。
  每个Observer对onNext等方法的处理逻辑也都是不一样的，还是举例来说明。
  比如MapObserver的onNext中，就是先将结果执行map中的逻辑，然后再传给downstream的onNext方法处理。
  比如ObserveOnObserver的onNext中，会将source的onNext方法抛到Worker中去处理，或者说抛到具体的某个线程中去处理。

关键源码解析

Observable.subscribe

在Observable的subscribe方法中，最终会调用到subscribeActual方法，subscribeActual方法是一个虚方法，会在子类中实现

    public final void subscribe(@NonNull Observer<? super T> observer) 
        Objects.requireNonNull(observer, "observer is null");
        try 
            observer = RxJavaPlugins.onSubscribe(this, observer);

            Objects.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");

            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;
        
    

    protected abstract void subscribeActual(@NonNull Observer<? super T> observer);

Observable.subscribeOn

subscribeOn最终创建的是ObservableSubscribeOn对象。

    public final Observable<T> subscribeOn(@NonNull Scheduler scheduler) 
        Objects.requireNonNull(scheduler, "scheduler is null");
        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<>(this, scheduler));
    

ObservableSubscribeOn.subscribeActual

subscribeOn这个方法是来决定订阅时执行的线程，因此在此处就需要抛到对应的线程去运行。

在subscribeActual中，最终会调用到Scheduler.scheduleDirect方法，在其中会将Observer中的逻辑放到新建的Worker中运行，即放到Scheduler设置的线程中运行。

    public void subscribeActual(final Observer<? super T> observer) 
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<>(observer);

        observer.onSubscribe(parent);

        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    

    public Disposable scheduleDirect(@NonNull Runnable run) 
        return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
    

    public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) 
        final Worker w = createWorker();

        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

        DisposeTask task = new DisposeTask(decoratedRun, w);

        w.schedule(task, delay, unit);

        return task;
    

ObservableSubscribeOn。SubscribeOnObserver.onNext

SubscribeOnObserver的onnext就只是调用了下上一个Observer的onNext方法。

此处有的读者可以会有点混乱，额外解释下。
Observable.subscribeOn：定义订阅时执行的线程
Observable.observeOn： 定义监听时执行的线程

        public void onNext(T t) 
            downstream.onNext(t);
        

Observable.observeOn

observeOn方法最终返回的是ObservableObserveOn对象。

    public final Observable<T> observeOn(@NonNull Scheduler scheduler) 
        return observeOn(scheduler, false, bufferSize());
    

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

ObservableObserveOn.subscribeActual

此处会创建Worker，即定义Observer要运行的线程，然后创建Observer传给source.subscribe方法。

    protected void subscribeActual(Observer<? super T> observer) 
        if (scheduler instanceof TrampolineScheduler) 
            source.subscribe(observer);
         else 
            Scheduler.Worker w = scheduler.createWorker();

            source.subscribe(new ObserveOnObserver<>(observer, w, delayError, bufferSize));
        
    

ObservableObserveOn.ObserveOnObserver.onNext

observeOn这个方法是来决定监听时执行的线程，因此会在Observer的方法中，将逻辑抛到对应的线程中执行。

此处最终会调用到worker.schedule方法，将逻辑放到对应的线程中去执行。

        public void onNext(T t) 
            if (done) 
                return;
            

            if (sourceMode != QueueDisposable.ASYNC) 
                queue.offer(t);
            
            schedule();
        

        void schedule() 
            if (getAndIncrement() == 0) 
                worker.schedule(this);