欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Android RxJava:一步步带你源码分析 RxJava

程序员文章站 2024-03-24 13:42:40
...

前言

  • Rxjava由于其基于事件流的链式调用、逻辑简洁 & 使用简单的特点,深受各大 Android开发者的欢迎。

Android RxJava:一步步带你源码分析 RxJava

如果还不了解RxJava,请看文章:Android:这是一篇 清晰 & 易懂的Rxjava 入门教程

  • 今天,我将为大家带来 源码分析:Rxjava的订阅流程,其为Rxjava使用的基本 & 核心,希望大家会喜欢。
    1. 本系列文章主要基于 Rxjava 2.0
    2. 接下来的时间,我将持续推出 AndroidRxjava 2.0 的一系列文章,包括原理、操作符、应用场景、背压等等 ,有兴趣可以继续关注Carson_Ho的安卓开发笔记!!

Android RxJava:一步步带你源码分析 RxJava

目录

Android RxJava:一步步带你源码分析 RxJava


1. RxJava简介

此处简单介绍RxJava

Android RxJava:一步步带你源码分析 RxJava

若还不了解RxJava,请看文章:Android:这是一篇 清晰 & 易懂的Rxjava 入门教程


2. 订阅流程 的使用

2.1 使用步骤

RxJava的订阅流程 使用方式 = 基于事件流的链式调用,具体步骤如下:

步骤1:创建被观察者(Observable)& 定义需发送的事件
步骤2:创建观察者(Observer) & 定义响应事件的行为
步骤3:通过订阅(subscribe)连接观察者和被观察者

2.2 实例讲解

// RxJava的链式操作
        Observable.create(new ObservableOnSubscribe<Integer>() {
        // 1. 创建被观察者(Observable) & 定义需发送的事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            // 2. 创建观察者(Observer) & 定义响应事件的行为
            // 3. 通过订阅(subscribe)连接观察者和被观察者
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "开始采用subscribe连接");
            }
            // 默认最先调用复写的 onSubscribe()

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "对Next事件"+ value +"作出响应"  );
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }

        });
    }
}
  • 运行结果

Android RxJava:一步步带你源码分析 RxJava


3. 源码分析

下面,我将根据 使用步骤 进行RxJava的源码分析:
步骤1:创建被观察者(Observable)& 定义需发送的事件
步骤2:创建观察者(Observer) & 定义响应事件的行为
步骤3:通过订阅(subscribe)连接观察者和被观察者

步骤1:创建被观察者(Observable)& 定义需发送的事件

  • 源码分析如下
/** 
  * 使用步骤1:创建被观察者(Observable)& 定义需发送的事件
  **/

  Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        })

/** 
  * 源码分析:Observable.create(new ObservableOnSubscribe<Integer>(){...})
  **/
    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {

    ...
      // 仅贴出关键源码

      return new ObservableCreate<T>(source);
      // 创建ObservableCreate类对象 ->>分析1
    // 注:传入source对象(即 我们手动创建的ObservableOnSubscribe对象)  
    }


  /** 
    * 分析1:new ObservableCreate<T>(source)
    **/

    public final class ObservableCreate<T> extends Observable<T> {
  // ObservableCreate类 = Observable的子类 

      ...
      // 仅贴出关键源码

        final ObservableOnSubscribe<T> source;

        // 构造函数
      // 传入了传入source对象 = 手动创建的ObservableOnSubscribe对象
        public ObservableCreate(ObservableOnSubscribe<T> source) {
            this.source = source;
        }

    /** 
      * 重点关注:复写了subscribeActual()
      * 作用:订阅时,通过接口回调 调用被观察者(Observerable) 与 观察者(Observer)的方法
      **/
        @Override
        protected void subscribeActual(Observer<? super T> observer) {

              // 1. 创建1个CreateEmitter对象(封装成1个Disposable对象)
          // 作用:发射事件
            CreateEmitter<T> parent = new CreateEmitter<T>(observer);

            // 2. 调用观察者(Observer)的onSubscribe()
            // onSubscribe()的实现 = 使用步骤2(创建观察者(Observer))时复写的onSubscribe()
            observer.onSubscribe(parent);

            try {
                // 3. 调用source对象的subscribe()
                // source对象 = 使用步骤1(创建被观察者(Observable))中创建的ObservableOnSubscribe对象 
                // subscribe()的实现 = 使用步骤1(创建被观察者(Observable))中复写的subscribe()->>分析2
                source.subscribe(parent);

            } catch (Throwable ex) {
                Exceptions.throwIfFatal(ex);
                parent.onError(ex);
            }
    }

  /** 
    * 分析2:emitter.onNext("1");
    * 此处仅讲解subscribe()实现中的onNext()
    * onError()、onComplete()类似,此处不作过多描述
    **/
    static final class CreateEmitter<T> extends AtomicReference<Disposable>
                                        implements ObservableEmitter<T>, Disposable {

        ...
        // 仅贴出关键代码

        // onNext()源码分析
        @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;
            }

            // 若无断开连接(调用Disposable.dispose()),则调用观察者(Observer)的同名方法 = onNext()
            // 观察者的onNext()的内容 = 使用步骤2中复写内容
            if (!isDisposed()) {
                observer.onNext(t);
            }
        }

        // onError()、onComplete()类似,此处不作过多描述
        // 特别说明:调用该2方法,最终都会自动调用dispose(),即断开观察者 & 被观察者的连接
        @Override
        public void onError(Throwable t) {
            if (t == null) {
                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
            }
            if (!isDisposed()) {
                try {
                    observer.onError(t);
                } finally {
                    dispose();
                }
            } else {
                RxJavaPlugins.onError(t);
            }
        }

        @Override
        public void onComplete() {
            if (!isDisposed()) {
                try {
                    observer.onComplete();
                } finally {
                    dispose();
                }
            }
        }
  • 步骤1总结

Android RxJava:一步步带你源码分析 RxJava

步骤2:创建观察者 & 定义响应事件的行为

  • 源码分析
/** 
  * 使用步骤2:创建观察者 & 定义响应事件的行为(方法内的创建对象代码)
  **/
  subscribe(new Observer<Integer>() {

              @Override
              public void onSubscribe(Disposable d) {
                  Log.d(TAG, "开始采用subscribe连接");
              }
              // 默认最先调用复写的 onSubscribe()

              @Override
              public void onNext(Integer value) {
                  Log.d(TAG, "对Next事件"+ value +"作出响应"  );
              }

              @Override
              public void onError(Throwable e) {
                  Log.d(TAG, "对Error事件作出响应");
              }

              @Override
              public void onComplete() {
                  Log.d(TAG, "对Complete事件作出响应");
              }

          });

/** 
  * 源码分析:Observer类
  **/
  public interface Observer<T> {
      // 注:Observer本质 = 1个接口

      // 接口内含4个方法,分别用于 响应 对应于被观察者发送的不同事件
        void onSubscribe(@NonNull Disposable d); // 内部参数:Disposable 对象,可结束事件
        void onNext(@NonNull T t);
        void onError(@NonNull Throwable e);
        void onComplete();
    }

/** 
  * 特别说明:Subscriber类
  * 定义:RxJava 内置的一个实现了 Observer 的抽象类
  * 作用:扩展Observer 接口 = 新增了2个方法 = 
  *      1. onStart():在还未响应事件前调用,用于初始化工作
  *      2. unsubscribe():用于取消订阅。在该方法被调用后,观察者将不再接收 & 响应事件
  *      注:调用该方法前,先使用 isUnsubscribed() 判断状态,确定被观察者Observable是否还持有观察者Subscriber的引用;若引用不能及时释放,就会出现内存泄露
  * 使用方式:与Observer使用几乎相同(实质上,Observer总是会先被转换成Subscriber再使用)
  **/
  Subscriber<String> subscriber = new Subscriber<Integer>() {

            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "开始采用subscribe连接");
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "对Next事件作出响应" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        };

步骤3:通过订阅(subscribe)连接观察者和被观察者

  • 源码分析
/** 
  * 使用步骤3:通过订阅(subscribe)连接观察者和被观察者 = subscribe()
  **/
  subscribe(new Observer<Integer>() {
              // 2. 通过通过订阅(subscribe)连接观察者和被观察者
              // 3. 创建观察者 & 定义响应事件的行为
              @Override
              public void onSubscribe(Disposable d) {
                  Log.d(TAG, "开始采用subscribe连接");
              }
              // 默认最先调用复写的 onSubscribe()

              @Override
              public void onNext(Integer value) {
                  Log.d(TAG, "对Next事件"+ value +"作出响应"  );
              }

              @Override
              public void onError(Throwable e) {
                  Log.d(TAG, "对Error事件作出响应");
              }

              @Override
              public void onComplete() {
                  Log.d(TAG, "对Complete事件作出响应");
              }

          });

/** 
  * 源码分析:Observable.subscribe(observer)
  * 说明:该方法属于 Observable 类的方法(注:传入1个 Observer 对象)
  **/  

  @Override
  public final void subscribe(Observer<? super T> observer) {

    ...
    // 仅贴出关键源码

    subscribeActual(observer);
    // 继续往下看:分析1

  }

/** 
  * Observable.subscribeActual(observer)
  * 说明:属于抽象方法,由子类实现;此处的子类 = 步骤1创建被观察者(Observable)时创建的ObservableCreate类
  * 即 在订阅时,实际上是调用了步骤1创建被观察者(Observable)时创建的ObservableCreate类里的subscribeActual()
  * 此时,你应该回头看上面的步骤1里的subscribeActual(),应该能理解RxJava的整个订阅流程了。
  **/
  protected abstract void subscribeActual(Observer<? super T> observer);
  • 总结
    Android RxJava:一步步带你源码分析 RxJava

4. 源码总结

  • 在步骤1(创建被观察者(Observable))、步骤2(创建观察者(Observer))时,仅仅只是定义了发送的事件 & 响应事件的行为;
  • 只有在步骤3(订阅时),才开始发送事件 & 响应事件,真正连接了被观察者 & 观察者
  • 具体源码总结如下

Android RxJava:一步步带你源码分析 RxJava


5. 特别注意:涉及多个被观察者(Observable)的发送事件顺序

  • 具体描述

Android RxJava:一步步带你源码分析 RxJava

  • 实例讲解
/** 
  * 存在涉及多个被观察者(Observable)的情况
  **/

    // 创建第1个被观察者(Observable1)
    Observable.create(new ObservableOnSubscribe<Integer>() {
                @Override
                public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                    emitter.onNext(1);
                    emitter.onNext(2);
                    emitter.onNext(3);
                }
            })
            // 使用flatMap操作符(内部会创建第2个被观察者(Observable2))
            .flatMap(new Function<Integer, ObservableSource<String>>() {
                @Override
                public ObservableSource<String> apply(Integer integer) throws Exception {
                    final List<String> list = new ArrayList<>();
                    for (int i = 0; i < 3; i++) {
                        list.add("我是事件" + integer + "拆分后的子事件" + i);
                        // 通过flatMap中将被观察者生产的事件序列先进行拆分,再将每个事件转换为一个新的发送三个String事件
                        // 最终合并,再发送给被观察者
                    }
                    return Observable.fromIterable(list);
                }

            })
            .subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.d(TAG, "开始采用subscribe连接");
                }
                // 默认最先调用复写的 onSubscribe()

                @Override
                public void onNext(String value) {
                    Log.d(TAG, "响应事件:"+ value   );
                }

                @Override
                public void onError(Throwable e) {
                    Log.d(TAG, "对Error事件作出响应");
                }

                @Override
                public void onComplete() {
                    Log.d(TAG, "对Complete事件作出响应");
                }
            });
            // 过程讲解
            // 调用顺序:先回调Observable2的subscribe(Observer) 、subscribeActual(Observer)、再回调Observable1的subscribe(Observer) 、subscribeActual(Observer)
            // Observable的发送顺序 = 先发送Observable1、再发送Observable2
  • 测试结果
    Android RxJava:一步步带你源码分析 RxJava

6. 总结

  • 本文主要对 RxJava2 中 的订阅流程进行了源码分析

  • 接下来的时间,我将持续推出 AndroidRxjava 2.0 的一系列文章,包括原理、操作符、应用场景、背压等等 ,有兴趣可以继续关注Carson_Ho的安卓开发笔记!!

Android RxJava:一步步带你源码分析 RxJava


请帮顶 / 评论点赞!因为你的鼓励是我写作的最大动力!

上一篇: jvm-class类文件结构(三)

下一篇: