RxJava入门与提高-Subject篇（2）

2017-11-09 本文已影响220人小玉1991

前言

欢迎继续收看《RxJava入门与提高》，上周出了第一篇 RxJava入门与提高（1）。
本文主要给大家补充一下上一篇遗留的Subject知识，没看过上一篇的同学、忘了上一章写什么的同学、还有其他同学，RxJava入门与提高（1）。温习一遍，俗话说，“书读百遍，奇异自见”，看多一遍是一遍，多多益善嘛。温习完的，请回来继续听讲。

Subject

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

它可以充当Observable；
它可以充当Observer；
它是Observable和Observer之间的桥梁；

接下来对这三个要点解释一下，但在解释之前，要先介绍一下Subject的种类， Subject是一个抽象类，不能通过new来实例化Subject，所以Subject有四个实现类，分别为AsyncSubject、BehaviorSubject、PublishSubject和ReplaySubject，每个实现类都有特定的“技能”，下面结合代码来介绍一下它们各自的“技能”。注意，所有的实现类都由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类型用作接受者（Observer）

至此，四种Subject类型已经介绍完毕，上文说过，Subject类型可以用作数据源（Observable），也可以用作接受源（Observer），或者两者之间的桥梁。介绍四种Subject类型，就是当做数据源（Observable）来介绍的。这里不在举例累赘。

但是需要注意，如果你把 Subject 当作一个 Observer（接受者）使用，不要从多个线程中调用它的onNext方法（包括其它的on系列方法），这可能导致同时（非顺序）调用，这会违反Observable协议，给Subject的结果增加了不确定性。

要避免此类问题，官方提出了“串行化”，你可以将 Subject 转换为一个 SerializedSubject ，类似于这样：（这个我没试验过，只是自己在资料中查到的方法）
```
SerializedSubject<String, Integer> ser = new SerializedSubject(publishSubject);
```

在实际开发中，用的多的是下边这种桥梁转发的方式。

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操作符的使用讲解。

欢迎继续收看：RxJava入门与提高-操作符篇（3）
作者：ZhangYushui
來源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

RxJava入门与提高-Subject篇（2）

前言

Subject

Subject的分类解析

Subject类型用作接受者（Observer）

总结

猜你喜欢

热点阅读