RxJava中几种不同的subject解释

在rxjava中，subject是即可作为观察者，也可以作为被观察者 。顾名思义

• subject作为被观察者（observable）
  即可作为observable向其他observable发送时间
• subject作为观察者（observer）
  即可作为observer接收其他observable发送的事件

一、observable的subject

看官方网文档，可以发现subject作为observable的时候主要有四个子类，分别是PublishSubject、BehaviorSubject、ReplaySubject、AsyncSubject。

• PublishSubject
  最常见的subject，当观察者订阅该subject，它会按照顺序发送事件给订阅的观察者。不会因为事件改变而改变顺序

  
  图片.png

• BehaviorSubject
  该类有创建时需要一个默认参数。并且在发送的时候会先发送离订阅最近的一个值，如果没有值的话就是用这个默认值，参考一下代码(参考)：

  
  图片.png

// 订阅之前，最近一个值为空，这里将发送默认值“default”给观察者
  BehaviorSubject<Object> subject = BehaviorSubject.create("default");
  subject.subscribe(observer);
  subject.onNext("one");
  subject.onNext("two");
  subject.onNext("three");

  // 订阅之前，最近一个值为“zero”，会先发送zero.因此观察者会接收到“one”、“two”、“three”
  BehaviorSubject<Object> subject = BehaviorSubject.create("default");
  subject.onNext("zero");
  subject.onNext("one");
  subject.subscribe(observer);
  subject.onNext("two");
  subject.onNext("three");

  // 观察者只能接收到onCompleted
  BehaviorSubject<Object> subject = BehaviorSubject.create("default");
  subject.onNext("zero");
  subject.onNext("one");
  subject.onCompleted();
  subject.subscribe(observer);

  // 观察者只能接收到onError
  BehaviorSubject<Object> subject = BehaviorSubject.create("default");
  subject.onNext("zero");
  subject.onNext("one");
  subject.onError(new RuntimeException("error"));
  subject.subscribe(observer);

• ReplaySubject
  该subject既有缓存事件，也就是当被一个观察者所订阅的时候，事件不会直接被消化。下个观察者订阅的时候可以再次接收到事件。

  
  图片.png

  ReplaySubject<Object> subject = ReplaySubject.create();
  subject.onNext("one");
  subject.onNext("two");
  subject.onNext("three");
  subject.onCompleted();

  // 两个观察者都会接收到事件
  subject.subscribe(observer1);
  subject.subscribe(observer2);

• AsyncSubject
  该subject只有在被观察者（subject）结束的时候（即执行了onCompleted）后观察者才能够接收到事件，并且也只有能接收到最后一个事件。

  
  图片.png

// 观察者不会接收到任何数据，因为subject没有结束（onCompleted）
  AsyncSubject<Object> subject = AsyncSubject.create();
  subject.subscribe(observer);
  subject.onNext("one");
  subject.onNext("two");
  subject.onNext("three");

  // 观察者只能接收到最后一件事件，即three事件
  AsyncSubject<Object> subject = AsyncSubject.create();
  subject.subscribe(observer);
  subject.onNext("one");
  subject.onNext("two");
  subject.onNext("three");
  subject.onCompleted();

二、observer的subject

这个日常应该会遇到的比较少，原理是subject订阅被观察者的时候，当被观察者发送事件到subject后，会发送到subject中注册的observer中。