Rxjava* 2.0 源码分析 2019-01-25

title: Rxjava* 2.0 源码分析

Rxjava 2.0 源码分析

参考：Android RxJava 2.0：手把手带你 源码分析RxJava https://www.jianshu.com/p/e1c48a00951a

版本：

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

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

源码分析：

/** 
  * 使用步骤1：创建被观察者(Observable)& 定义需发送的事件
  **/
 Observable.create(new ObservableOnSubscribe<Integer>() {
            // 2. 在复写的subscribe（）里定义需要发送的事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }
        // 内部类 实现 subscribe 的方法 ObservableOnSubscribe 对象
        );


/** 
  * 源码分析：Observable.create(new ObservableOnSubscribe<Integer>(){...}）
  **/
 
  @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));
        //   创建 Observable 类 子对象 ->>分析1
        // 注：传入source对象（即我们手动创建的ObservableOnSubscribe对象）
        
    }
    
 /** 
    * 分析1：RxJavaPlugins.onAssembly(new ObservableCreate<T>(source))
    **/

    public static <T> Observable<T> onAssembly(Observable<T> source) {
        Function<Observable, Observable> f = onObservableAssembly;
        // 实例化 Function 对象 ，对 Function 对象赋值
        //第一次 未赋值 f=null
        //f 赋值后  apply(f, source); ->> 分析2
        if (f != null) {
            return apply(f, source);
        }
        
        return source;
        //第一次 未赋值 返回的是 ObservableCreate 对象 ->> 分析4 
      
    }
    
    
/**
 *分析2 apply(f, source)
 */
    
   static <T, R> R apply(Function<T, R> f, T t) {
        try {
            return f.apply(t);
            //返回 R 对象  继续查看代码 ->> 分析3
        } catch (Throwable ex) {
            throw ExceptionHelper.wrapOrThrow(ex);
        }
    } 
  
     
/**
 *分析3 f.apply(t)
 */ 
  public interface Function<T, R> {
    /**
     * Apply some calculation to the input value and return some other value.
     * @param t the input value
     * @return the output value
     * @throws Exception on error
     */
    R apply(T t) throws Exception;
    // 如果 function 有实例 ，会调用其 apply 方法 ，返回 R 对象
}
  
 
     
/**
 *分析4  new ObservableOnSubscribe<T> 对象
 
 ObservableOnSubscribe 对象有 两个 静态内部类 和一个方法
 */    
 
 public final class ObservableCreate<T> extends Observable<T> {
 //ObservableCreate 为Observable 的子类
    final ObservableOnSubscribe<T> source;
   // 仅贴出关键源码
   
     // 构造函数
     // 传入了传入source对象 = 手动创建的ObservableOnSubscribe对象
    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }
   
   //其内部的 静态内部类 1 其为 序列化的类
  /** emitter.onNext("1");
    * 此处仅讲解subscribe（）实现中的onNext（）
    * onError（）、onComplete()类似，此处不作过多描述*/
static final class CreateEmitter<T>  extends  AtomicReference<Disposable>
    implements ObservableEmitter<T>, Disposable {
    
    
    private static final long serialVersionUID = -3434801548987643227L;
    
    final Observer<? super T> observer;
    //构造方法  传入 的 是观察者对象
    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;
            }
     // 若无断开连接（调用Disposable.dispose()），则调用观察者（Observer）的同名方法 = onNext（）
     // 观察者的onNext（）的内容 = 使用步骤2中复写内容
            if (!isDisposed()) {
                observer.onNext(t);
            }
        }

        @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();
                }
            }
        }
    
    
  //其他   
  @Override
    public void dispose() {
        DisposableHelper.dispose(this);
    }

    @Override
    public boolean isDisposed() {
        return DisposableHelper.isDisposed(get());
    }
    
   }
   
     //其内部的 静态内部类 2 其为 序列化的类
 static final class SerializedEmitter<T>    extends AtomicInteger
    implements ObservableEmitter<T> {
    
        
    }
   
   //内部的 主要方法  翻译为 实际订阅
     /** 
      * 重点关注：复写了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（）->>分析4

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

步骤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) {

 
    // 仅贴出关键源码
   
 //   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;
        }
  }

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


//对观察者 进行了多次处理  ，f 为空 则用原来的 ，如果 不为空 则走 apply（..,..,..）方法
  public static <T> Observer<? super T> onSubscribe(Observable<T> source, Observer<? super T> observer) {
        BiFunction<Observable, Observer, Observer> f = onObservableSubscribe;
        if (f != null) {
            return apply(f, source, observer);
        }
        return observer;
    }
image

4. 源码总结

  • 在步骤1（创建被观察者（Observable））、步骤2（创建观察者（Observer））时，仅仅只是定义了发送的事件 & 响应事件的行为；
  • 只有在步骤3（订阅时），才开始发送事件 & 响应事件，真正连接了被观察者 & 观察者

  • 具体源码总结如下


    image

5. 特别注意：涉及多个被观察者（Observable）的发送事件顺序

image

实例讲解：

/** 
  * 存在涉及多个被观察者（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

测试结果：


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

推荐阅读更多精彩内容