#1 Subject 的基本概念

通常我们碰到的Observable 和 observer 都是一一对应的，多个observers 即使订阅相同的observable,它们之间也是独立的，不会共享数据或者事件，比如下例

示例1·

var observable = Rx.Observable.interval(1000).take(3);

# 观察者A
var observerA = {
  next: function(x) { console.log('观察者A ' + x);},
  error: function(err) { console.log(err); },
  complete: function() { console.log('A完成'); }
}

observable.subscribe(observerA); # 观察者A订阅observable 相当于执行函数

# 观察者B
var observerB = {
  next: function(x) { console.log('观察者B ' + x);},
  error: function(err) { console.log(err); },
  complete: function() { console.log('B完成'); }
}

setTimeout(() => {
  observable.subscribe(observerB); # 观察者B订阅相同的observable 相当于执行函数
}, 2000)

打印结果

"观察者A 0"
"观察者A 1"
"观察者B 0"
"观察者A 2"
"A完成"
"观察者B 1"
"观察者B 2"
"B完成"

可以看出观察者B虽然2秒后订阅相同的observable,但是完全不受观察者A的影响，仍然从0开始然后结束。

但是有时候我们希望流(即Observable)的数据或者事件在多个观察者之间是共享的，这个时候 Subject 就该登场了

Subject 的特点，它既是一个Observable(拥有所有Observable该拥有的操作符和方法，以及能够被别的观察者订阅的能力)，又是Observer(拥有next, error, complete等方法)，就是这么强悍的能力，可以把它看成一个数据的桥梁

Subject 的基本实现

上面的示例中我们看到因为订阅相当于函数执行，上面订阅了2次，所以函数执行了2次，如果想要函数只执行1次，我们可以通过一个桥梁，将所有订阅者收集到一个集合里面，然后再订阅1次，即函数只执行一次

示例2

var observable = Rx.Observable.interval(1000).take(3);

var bridgeObserver = {
  next: function(x) {
    this.observers.forEach(o => o.next(x));
  },
  error: function(err) {
    this.observers.forEach(o => o.error(err));
  },
  complete: function() {
    this.observers.forEach(o => o.complete());
  },
  observers: [],  // 观察者集合
  addObserver: function(observer) { // 将观察者添加到观察者集合中
    this.observers.push(observer)
  }
}

var observerA = {
  next: function(x) { console.log('观察者A ' + x);},
  error: function(err) { console.log(err); },
  complete: function() { console.log('A完成'); }
}

observable.subscribe(bridgeObserver); // 只订阅一次 所以相当于函数只执行一次

bridgeObserver.addObserver(observerA); // observerA添加到观察者集合中

var observerB = {
  next: function(x) { console.log('观察者B ' + x);},
  error: function(err) { console.log(err); },
  complete: function() { console.log('B完成'); }
}

setTimeout(() => {
  bridgeObserver.addObserver(observerB); // observerB添加到观察者集合中
}, 2000)

打印结果

Console Run  Clear
"观察者A 0"
"观察者A 1"
"观察者B 1"
"观察者A 2"
"观察者B 2"
"A完成"
"B完成"

可以看出观察者B共享了A中的事件，并没有从头开始执行

将上面的桥梁bridgeObserve换成 subject, addObserver 换成 subscribe 即为我们通常所见的Subject的基本骨架了

即

var observable = Rx.Observable.interval(1000).take(3);

var subject = {  // **`bridgeObserve`**换成 **`subject`**, 
  next: function(x) {
    this.observers.forEach(o => o.next(x));
  },
  error: function(err) {
    this.observers.forEach(o => o.error(err));
  },
  complete: function() {
    this.observers.forEach(o => o.complete());
  },
  observers: [],
  subscribe: function(observer) { // **`addObserver`** 换成 **`subscribe`** 
    this.observers.push(observer)
  }
}

var observerA = {
  next: function(x) { console.log('观察者A ' + x);},
  error: function(err) { console.log(err); },
  complete: function() { console.log('A完成'); }
}

observable.subscribe(subject); // subject作为observer的功能 能够订阅observable

subject.subscribe(observerA); // subject作为observable的功能 能够被其它observer订阅

var observerB = {
  next: function(x) { console.log('观察者B ' + x);},
  error: function(err) { console.log(err); },
  complete: function() { console.log('B完成'); }
}

setTimeout(() => {
  subject.subscribe(observerB)  // subject作为observable的功能 能够被其它observer订阅
}, 2000)

最后Rx为我们提供了这个桥梁称之为 Subject, 所以上面的代码等同于

var observable = Rx.Observable.interval(1000).take(3);

// var subject = {
//   next: function(x) {
//     this.observers.forEach(o => o.next(x));
//   },
//   error: function(err) {
//     this.observers.forEach(o => o.error(err));
//   },
//   complete: function() {
//     this.observers.forEach(o => o.complete());
//   },
//   observers: [],
//   subscribe: function(o) {
//     this.observers.push(o)
//   }
// }

// RxJS给我们提供了这样一个接口
var subject = new Rx.Subject(); 

var observerA = {
  next: function(x) { console.log('观察者A ' + x);},
  error: function(err) { console.log(err); },
  complete: function() { console.log('A完成'); }
}

observable.subscribe(subject);

subject.subscribe(observerA)

var observerB = {
  next: function(x) { console.log('观察者B ' + x);},
  error: function(err) { console.log(err); },
  complete: function() { console.log('B完成'); }
}

setTimeout(() => {
  subject.subscribe(observerB)
}, 2000)