Subject = Observable + Observer

subject是个神奇的对象，他可以是一个Observable（观察）同时也可以是一个Observer（观察者）：它作为链接这两个世界的桥梁。一个Subject可以订阅一个Observable，就像一个观察者，并且它可以发射新的数据，就像一个Observable。很明显，作为Obserable，观察者们或者其它Subject都可以订阅它。

一旦Subhect订阅了Observable，他将会触发Observable开始发射。如果原始的Observable是“冷”的，这将会对订阅一个“热”的Observable变量产生影响。

RxJava提供了四种不同的Subject：

• PublishSubject （发布）
• BehaviorSubject （行为/反应）
• ReplaySubject （重播）
• AsyncSubject （异步）

PublishSubject

Publish是Subject的一个基础子类。让我们看看用PublishSubject实现传统的Observable

  PublishSubject<String> stringPublishSubject = PublishSubject.create();
  Subscription subscriptionPrint = stringPublishSubject.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
    System.out.println("Observable completed");
}

@Override
public void onError(Throwable e) {
    System.out.println("Oh,no!Something wrong happened!");                
}

@Override
public void onNext(String message) {
    System.out.println(message);
}
  });
    stringPublishSubject.onNext("Hello World");

在刚才的例子中，我们创建一个PulishSubject，用create(）发射一个S听值，然后我们订阅PublishSubject。此时，没有数据要发送，因此我们的观察者只能等待，没有阻塞线程，也没有资源消耗。
最后一行代码展示了手动发射字符串“Hello World”,它触发了观察者的onNext()方法，让我们在控制台打印出“Hello World”信息。
让我们看一个更复杂的例子。话说我们有一个private声明的Observable，外部不能访问。Observable在它生命周期内发射值，我们不用关心这些值，我们只关心他们的结束。
首先，我们创建一个新的PublishSubject来响应它的onNext()方法，并且外部也可以访问它。

final PublishSubject<Boolean> subject = PublishSubject.create();
subject.subscribe(new Observer<Boolean>() {
@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {

}

@Override
public void onNext(Boolean aBoolean) {
    System.out.println("Observable Completed");
}
});

然后，我们创建“私有”的Observable，只有subject才可以访问的到。

 Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
    for (int i = 0; i < 5; i++) {
        subscriber.onNext(i);
    }
    subscriber.onCompleted();
}
}).doOnCompleted(new Action0() {
@Override
public void call() {
    subject.onNext(true);
}
}).subscribe();

observable.create()方阿飞包含了我们for循环，发射数字。doOnCompleted()方法指定当Observable结束时要做什么事情；在subject上发射true。最后，我们订阅了Observable。很明显，空的subscribe（）调用仅仅是为了开启Observable，而不用管已发出的任何值，也不用管完成事件或者错误事件。为了这个例子我们需要它像这样。
在这个例子中，我们创建了一个可以连接Observables并且同时可被观测的实体。当我们想为公共资源创建独立、抽象或更易观测的点时，这是极其有用的。

BehaviorSubject

简单的说，BehaviorDSubject会首先向他的订阅这发送截至订阅前最新的一个数据对象（或初始值），然后正常发送订阅后的数据流。
BehaviorSubject<Integer> behaviorSubject = BehaviorSubject.create(1);