Observer (观察者)

什么是观察者？-前面的博客我们讲了Observable可观察对象，里面我们讲了一个上自习放风的案例，其实那些同学们就是观察者。

观察者是可观察对象(Observable) 所发送数据的消费者。观察者只是一组回调函数的集合，每个回调函数对应一种 Observable 发送的通知类型：next、error 和 complete 。

代码说明一个典型的观察者对象：

var observer = {
    next: x => console.log(x + '来了快安静 '),
    err => console.error('不好意思我没有看到'),
    () => console.log('下课了不用放风了')
}

注意：此处的next、error 和 complete与可观察对象上的并不是一回事，这里的这个指的是观察者的，是另外一个新东西不要混淆。

其实你可以这样想！那些同学(观察者)他们有一系列的行为(一组回调函数的集合)：

老师来了->放风的同学看到了->通知大家->大家很安静(next)
放学了->放风的同学下线放风职责结束->大家不需要接受消息了(complete)
老师来了->放风的同学没有看到(没有到通知大家感到抱歉)，老师通知自己来了->小明被抓到玩手机，手机没收；小丽被抓到看小说，小说没收(error)

要使用观察者，需要把它提供给 Observable 的subscribe方法进行订阅才可以：

observable.subscribe(observer)

观察者不过是三个回调函数组成的对象，每个回调函数分别对应可观察对象的通知类型。

RxJS 中的观察者也可能是部分的。如果你没有提供某个回调函数，Observable 的执行也会正常运行，只是某些通知类型会被忽略，因为观察者中没有没有相对应的回调函数。

下面的示例是没有 complete 回调函数的观察者：

var observer = {
  next: x => console.log('Observer got a next value: ' + x),
  error: err => console.error('Observer got an error: ' + err),
};

当订阅 Observable(可观察对象) 时，你可能只提供了一个回调函数作为参数，而并没有将其附加到观察者对象上，这也是可以的，例如我们之前案例那样：

observable.subscribe(x => console.log('Observer got a next value: ' + x));

这时在 observable.subscribe 内部，它会创建一个观察者对象并使用第一个回调函数参数作为 next的处理方法。

当然也可以通过将三个函数作为参数提供三种回调:

observable.subscribe(
  x => console.log('Observer got a next value: ' + x),
  err => console.error('Observer got an error: ' + err),
  () => console.log('Observer got a complete notification')
);

完整代码实例

var cont = 0
// 创建可观察对象
var observable = Rx.Observable.create(function subscribe (observer) { // 执行 Observables
    var intervalID = setInterval(() => {
        cont++
        try {
            let random = Math.floor(Math.random() * 10)
            if (random < 1) { // 被老师发现是随机的
                throw new Error('没有发现老师错误')
            } else {
                observer.next('老师来了快安静')
            }
        } catch (err) {
            console.error(err)
            console.log('老师：我发现你们了---批评一顿，无法放学')
            observer.error(err) // 如果捕获到异常会发送一个 JavaScript 错误 或 异常
        }
    }, 2000) // 老师每隔2s种来一次
    setTimeout(() => { // 异步执行
        observer.complete()
    }, 10000) // 10s后放学
    // 提供取消和清理 interval 资源的方法
    return function unsubscribe() {
        cont = 0
        clearInterval(intervalID)
    }
})
// 订阅
var subscription = observable.subscribe({ // 观察者-同学小明
    next: x => {
        console.log('通知次数' + cont + ':' + x)
        console.log('小明停止了玩手机')
        console.log('-----------')
    },
    error: function (err) {
        console.error('小明被抓到玩手机，手机没收')
        this.unsubscribe() // 清除
        console.log(cont)
    },
    complete: function () {
        console.log('放学了，不需要观察了')
        this.unsubscribe()
        console.log(cont)
    }
})