美文网首页Android-RxJava服务治理 Rxjava相关
我所理解的RxJava——上手其实很简单(二)

我所理解的RxJava——上手其实很简单(二)

作者: Weavey | 来源:发表于2016-08-31 17:05 被阅读23488次

    前言

    欢迎继续收看《我所理解的RxJava--上手其实很简单(二)》,上周出了第一篇,各位程序猿大大的反应还不错,生平第一篇博文能获得大家的肯定,内心肯定是非常开心的,也坚定了我继续写下去的信念,总之,先谢谢大家送出的鱼丸...哦不,是收藏。好了,题外话不多说,进入这一篇的主题,本文主要给大家补充一下上一篇遗留的Subject知识,没看过上一篇的同学、忘了上一章写什么的同学、还有其他同学,请进入时光机:我所理解的RxJava——上手其实很简单(一)温习一遍,俗话说,“书读百遍,奇异自见”,看多一遍是一遍,多多益善嘛。温习完的,请回来继续听讲。


    关于Subject

    关于Subject,官方文档的解释是这样的:Subject可以看成是一个桥梁或者代理,在某些ReactiveX实现中(如RxJava),它同时充当了Observer和Observable的角色。因为它是一个Observer,它可以订阅一个或多个Observable;又因为它是一个Observable,它可以转发它收到(Observe)的数据,也可以发射新的数据。从官方解释中,我提取出三个要点:

    1. 它可以充当Observable;
    2. 它可以充当Observer;
    3. 它是Observable和Observer之间的桥梁;

    接下来对这三个要点解释一下,但在解释之前,要先介绍一下Subject的种类, Subject是一个抽象类,不能通过new来实例化Subject,所以Subject有四个实现类,分别为AsyncSubjectBehaviorSubjectPublishSubjectReplaySubject,每个实现类都有特定的“技能”,下面结合代码来介绍一下它们各自的“技能”。注意,所有的实现类都由create()方法实例化,无需new,所有的实现类调用onCompleted()onError(),它的Observer将不再接收数据;


    Subject的分类解析

    • AsyncSubject
      Observer会接收AsyncSubject的```onComplete()``之前的最后一个数据,如果因异常而终止,AsyncSubject将不会释放任何数据,但是会向Observer传递一个异常通知。示例代码如下:
            AsyncSubject<String> asyncSubject = AsyncSubject.create();
            asyncSubject.onNext("asyncSubject1");
            asyncSubject.onNext("asyncSubject2");
            asyncSubject.onNext("asyncSubject3");  
            asyncSubject.onCompleted();
            asyncSubject.subscribe(new Observer<String>() {
                @Override
                public void onCompleted() {
    
                    LogUtil.log("asyncSubject onCompleted");  //输出 asyncSubject onCompleted
                }
    
                @Override
                public void onError(Throwable e) {
    
                    LogUtil.log("asyncSubject onError");  //不输出(异常才会输出)
                }
    
                @Override
                public void onNext(String s) {
    
                    LogUtil.log("asyncSubject:"+s);  //输出asyncSubject:asyncSubject3
                }
            });
    

    以上代码,Observer只会接收asyncSubject的onCompleted()被调用前的最后一个数据,即“asyncSubject3”,如果不调用onCompleted(),Subscriber将不接收任何数据。

    • BehaviorSubject
      Observer会接收到BehaviorSubject被订阅之前的最后一个数据,再接收其他发射过来的数据,如果BehaviorSubject被订阅之前没有发送任何数据,则会发送一个默认数据。(注意跟AsyncSubject的区别,AsyncSubject要手动调用onCompleted(),且它的Observer会接收到onCompleted()前发送的最后一个数据,之后不会再接收数据,而BehaviorSubject不需手动调用onCompleted(),它的Observer接收的是BehaviorSubject被订阅前发送的最后一个数据,两个的分界点不一样,且之后还会继续接收数据。)示例代码如下:
        BehaviorSubject<String> behaviorSubject = BehaviorSubject.create("default");
        behaviorSubject.onNext("behaviorSubject1");
        behaviorSubject.onNext("behaviorSubject2");
            behaviorSubject.subscribe(new Observer<String>() {
                @Override
                public void onCompleted() {
    
                    LogUtil.log("behaviorSubject:complete");
                }
    
                @Override
                public void onError(Throwable e) {
    
                    LogUtil.log("behaviorSubject:error");
                }
    
                @Override
                public void onNext(String s) {
    
                    LogUtil.log("behaviorSubject:"+s);
                }
            });
    
            behaviorSubject.onNext("behaviorSubject3");
            behaviorSubject.onNext("behaviorSubject4");
    

    以上代码,Observer会接收到behaviorSubject2、behaviorSubject3、behaviorSubject4,如果在behaviorSubject.subscribe()之前不发送behaviorSubject1、behaviorSubject2,则Observer会先接收到default,再接收behaviorSubject3、behaviorSubject4。

    • PublishSubject
      PublishSubject比较容易理解,相对比其他Subject常用,它的Observer只会接收到PublishSubject被订阅之后发送的数据。示例代码如下:
      PublishSubject<String> publishSubject = PublishSubject.create();
      publishSubject.onNext("publishSubject1");
      publishSubject.onNext("publishSubject2");
      publishSubject.subscribe(new Observer<String>() {
             @Override
             public void onCompleted() {
                 
             }
      
             @Override
             public void onError(Throwable e) {
      
             }
      
             @Override
             public void onNext(String s) {
                 LogUtil.log("publishSubject observer1:"+s);
             }
         });
      publishSubject.onNext("publishSubject3");
      publishSubject.onNext("publishSubject4");
      

    以上代码,Observer只会接收到"behaviorSubject3"、"behaviorSubject4"。

    • ReplaySubject
      ReplaySubject会发射所有数据给观察者,无论它们是何时订阅的。也有其它版本的ReplaySubject,在重放缓存增长到一定大小的时候或过了一段时间后会丢弃旧的数据。示例代码如下:
    ReplaySubject<String>replaySubject = ReplaySubject.create(); //创建默认初始缓存容量大小为16的ReplaySubject,当数据条目超过16会重新分配内存空间,使用这种方式,不论ReplaySubject何时被订阅,Observer都能接收到数据
    //replaySubject = ReplaySubject.create(100);//创建指定初始缓存容量大小为100的ReplaySubject
    //replaySubject = ReplaySubject.createWithSize(2);//只缓存订阅前最后发送的2条数据 
     //replaySubject=ReplaySubject.createWithTime(1,TimeUnit.SECONDS,Schedulers.computation());  //replaySubject被订阅前的前1秒内发送的数据才能被接收     
    replaySubject.onNext("replaySubject:pre1");
    replaySubject.onNext("replaySubject:pre2");
    replaySubject.onNext("replaySubject:pre3");
    replaySubject.subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                    LogUtil.log("replaySubject:" + s);
            }
        });
    replaySubject.onNext("replaySubject:after1");
    replaySubject.onNext("replaySubject:after2");
    

    以上代码,由于情况比较多,注释也已解释的相当清楚,就不对输出结果一一表述了,有疑问的自行copy代码去测试一下。至此,四种Subject类型已经介绍完毕,但是需要注意,如果你把 Subject 当作一个 Subscriber 使用,不要从多个线程中调用它的onNext方法(包括其它的on系列方法),这可能导致同时(非顺序)调用,这会违反Observable协议,给Subject的结果增加了不确定性。要避免此类问题,官方提出了“串行化”,你可以将 Subject 转换为一个 SerializedSubject ,类似于这样:

    SerializedSubject<String, Integer> ser = new SerializedSubject(publishSubject);
    

    要点解答

    接下来,我们继续前面提出的问题,为什么说Subject既可充当Observable,又可充当Observer,是它们两个之间的桥梁呢?经过前面的例子,也许有些人已经大概理解了,不理解的且听我细细道来。首先,从理论上讲,Subject继承了Observable,又实现了Observer接口,所以说它既是Observable又是Observer,完全合理。从实际应用上讲,Subject也能实现Observable和Observer相同的功能,口说无凭,我们还是通过代码来证实比较有说服力。

    • 创建Observable并发射数据:
        Observable.create(new Observable.OnSubscribe<String>() {
                @Override
                public void call(Subscriber<? super String> subscriber) {
    
                    subscriber.onNext("I'm Observable");
                    subscriber.onCompleted();
                }
            });
    

    用Subject实现为:

    PublishSubject<String> publishSubject = PublishSubject.create();
    publishSubject.onNext("as Observable");
    publishSubject.onCompleted();
    
    • 创建Observer订阅Observable并接收数据:
    mObservable.subscribe(new Observer<String>() {
            @Override
             public void onCompleted() {
                    
            }
    
            @Override
            public void onError(Throwable e) {
    
            }
    
            @Override
            public void onNext(String s) {
      
                    //接收数据
            }
    });
    

    用Subject实现为:

         publishSubject.subscribe(new Observer<String>() {
                @Override
                public void onCompleted() {
                    
                }
    
                @Override
                public void onError(Throwable e) {
    
                }
    
                @Override
                public void onNext(String s) {
    
                }
            });
    

    也许有人会问,不是说Subject也可以作为Observer,不能把Subject当作Observer传入subscribe()中吗?回答是:当然可以!就象这样:

    PublishSubject<String> publishSubject = PublishSubject.create();
    Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
    
                subscriber.onNext("as Observer"); 
                subscriber.onCompleted();
            }
    }).subscribe(publishSubject);
    

    有没有发现问题?publishSubject没有重写onNext()方法啊,在哪接收的数据?这就是前面说的“桥梁”的问题了,尽管把Subject作为Observer传入subscribe(),但接收数据还是要通过Observer来接收,借用Subject来连接Observable和Observer,整体代码如下:

    PublishSubject<String> publishSubject = PublishSubject.create();
         Observable.create(new Observable.OnSubscribe<String>() {
                @Override
                public void call(Subscriber<? super String> subscriber) {
    
                    subscriber.onNext("as Bridge");
                    subscriber.onCompleted();
                }
            }).subscribe(publishSubject);
    
            publishSubject.subscribe(new Observer<String>() {
                @Override
                public void onCompleted() {
                    
                }
    
                @Override
                public void onError(Throwable e) {
    
                }
    
                @Override
                public void onNext(String s) {
    
                    LogUtil.log("subject:"+s); //接收到 as Bridge
                }
            });
        ```
    没错,这很桥梁!
    ***
    ##总结
    关于Subject,到此就介绍完了。也许你会跟我一样困惑,为什么又要多个Subject出来,除了有几个特定功能之外,其他所有的一切,Observable和Observer也都有,而且写法上也没有原来的简便。确实如此,对于几个特定功能,我也还想不到有什么应用场景,至少我还没发现有什么场景必须得用Subject来实现不可,那么问题又来了,我为什么要花这么大篇幅来介绍Subject,理由有三。其一,既然官方推出Subject,必有其道理,还没遇到不代表以后不会遇到,更不能代表你不会遇到这样的应用场景;其二,“一千个读者有一千个哈姆雷特”,我所看到的并不是全部,也许你会发掘出更有意思的东西可不是?其三,我可不想当你看完我所有关于RxJava的文章,自信已上手RxJava,当有人跟你提起Subject的时候,你一脸茫然不知道Subject是什么东西,岂不哀哉?所以呢,介绍一下Subject还是很有意义的,最起码学了比没学好,“养兵千日用兵一时”,知识不嫌多,突然哪天就用上了呢。对于Subject的理解,有异议的欢迎底下评论,一起交流进步。下一篇文章,进入RxJava操作符的使用讲解。
    ***
    本猿已转行,勿念。

    相关文章

      网友评论

      • 高级组装搬运工吴哥:subject的用法讲得很好很清晰
      • messiGao:写的不错。书读百遍,奇异自见 这个改下,写错了。
      • 般若同学:没错,这很桥梁,哈哈
      • glong:书读百遍,其义自见
      • itstrongs:写的不错,那么问题来了,三呢?
      • 2d51cabea192:我也是看了几篇文章都看不懂,当看到你这篇文章的时候我才豁然开朗
      • 风影_638f:啥时候出3?
      • dc24fc83b9b6:最后的例子写的好像有点BUG,Observable订阅PublicSubject的时候,publicSubject还没有订阅Observer呢,怎么会接收到数据呢?
        RyanYans32:确实是个BUG
        留存的情缘: @老坛酸红烧肉 估计这里就是一个桥梁实现observable到observer
      • SoloHo:我的理解,Subject是响应式编程的核心,响应式编程就是构建关系,然后当某个事件触发,另外一个关系同时发生改变。参考Android官方的LiveData,会更容易明白响应式编程,否则RxJava的使用只是在使用操作符而已,并非是响应式编程。
      • 年轻就应该奔跑:讲讲 这些 用法吧 flatMap filter map
      • tyh520life:讲讲 这些 用法吧 flatMap filter map
      • 小小N:专程登录给您好评,总结得很到位,简明易懂。
      • 易浪:Nice
      • psj_psj:看到这,我有个应用场景,但是不知道怎么实现,就是两个文件(两个class)之间发送和接收数据,这个咋实现啊?
      • yejg:加油!不要停下来。。。
      • 0795bf4f06ba:期待你下一篇呢
      • MambasJi:很赞,请问会有三吗
      • itlong:#不错 通俗易懂 :smiley:
      • 边城狂人:貌似 Ovservable 只能有一个订阅者,Subject 可以解决多个订阅者的问题……我只是听到有这么一说,没去研究。
        doc__wei: Ovservable 可以有很多个订阅者啊,,,这里使用create的时候就创建了一个subscriber,而且后面还有创建observer
      • Kwoner:大神讲的很好,受用。
      • Wang_Guan:三呢三呢?几时出?
      • T9的第三个三角:坐等,看了几次Rxjava都看不下去
      • 你需要一台永动机: :clap: :clap: :clap: 给你赞一个
      • 健康早餐:我在用RxBus的时候,需要使用Subject的几个实现类,来调整接受数据的机制
        Mansoul:是的RxBus中会用到Subject当时看到是懵逼的,哈哈现在回头看豁然开朗
        Weavey:@帅气的昵称啊123 是什么样的需求呢?
      • 1934d32ccddc:期待更多帮助新手小白的,表示很多都看不懂,大概因为我有点古代人
        Weavey: @墨脱明光 多看几遍说不定就懂了哦,我也是看了好几遍才明白,有问题欢迎一起交流😊😊😊
      • 世锋日上:哎哟,不错哦
      • 四会歌神陈子豪:大神什么时候出(三)啊,等不及了
      • Wower_Lau:作为小白菜的我,真是受益匪浅呀,感谢大神分享! :clap: :clap:
        Weavey:@Wower_Lau 让大神见笑了 :pray:

      本文标题:我所理解的RxJava——上手其实很简单(二)

      本文链接:https://www.haomeiwen.com/subject/igetettx.html