可观察对象和观察者模式的差别:

• 可观察者对象不启动流媒体项目(streaming items),直到至少有一个观察者订阅(Observer)它(这句话很重要)
• 和迭代器类似，一个可观察者对象当序列(sequence)完成时能发送信号(signal)
• 使用可观察者，我们能够声明如何响应它发出的序列元素，而不是响应当单独的个体元素
• 我们可以高效的复制，转换和查询序列，这些操作对所有序列元素都有效

Observables && Observer

有很多种方式创建Observables对象，使用 create 方法(在Rxjs中称之为操作符)是最明显的方式。

Rx.Observable接收一个以Observer作为参数的回调，这个回调函数定义可观察对象如何发出值(emit values).

var observable = Rx.Observable.create(function(observer) {
  observer.onNext('James');
  observer.onNext('Jordan');
  observer.onNext('Durant');
  observer.onCompleted(); // 完成
})

当订阅这个可观察对象，它将调用 onNext 方法， 生成3个字符串，然后调用 onCompleted 发出信号，表示序列完成。我们使用 Observers 来订阅 Observable

Observers

Observers订阅可观察对象，当可观察对象发生变化，观察者调用相关的方法，它有3个方法:

• onNext: 相当于观察者模式中的 ``Update 方法
• onError: 可观察对象发生错误时调用，它调用之后，再调用onNext是无效的
• onCompleted： 发出信号，没有更多数据了

创建观察者:

var observer = Rx.Observer.create(
  function onNext(x) { console.log('next: ' + x);},
  function onError(err) { console.log('Error: ' + err); },
  function onCompleted() { console.log('Done'); }
);

Rx.Observer.create接收函数 onNext，onError，onCompleted， 返回一个Observer实例，这些方法是可选的。比如有些无限序列(比如用户一直点击鼠标)，onCompleted是永远不会被调用的，如果有信心不发生错误，onError也是可以不用写的。

Observable ajax请求

使用可观察对象来封装一个ajax请求,返回一个可观察对象实例：

function getUrl(url) {
  return Rx.Observable.create(function(observer) {
    var xhr = new XMLHttpRequest();
    xhr.open('GET', url);

    xhr.onload = function() {
      if (req.status === 200) {
        observer.onNext(xhr.response);
        observer.onCompleted();
      } else {
        observer.onError(new Error(xhr.statusText));
      }
    };

    xhr.onerror = function() {
      observer.onError(new Error('未知错误'));
    };

    xhr.send();
  })
}

光有可观察对象，没有观察者，流是不会触发的，下面是观察者:

var test = getUrl('https://randomuser.me/api/')

// 订阅一个观察者
test.subscribe(
  function onNext(x) {console.log('Result: ' + x)},
  function onError(err) {console.log('Error: ' + err)},
  function onCompleted() {console.log('done')}
);

此处我们可以发现，我们没有通过Rx.Observer.create显式的创建一个Observer, 而是通过 subscribe操作符来创建Observer

操作符

在Rxjs中，转换或查询序列方法被称之为 operators（操作符）。上面的ajax封装可以使用已经存在的rxjs-dom库来完成。

// 首先先引入rxjs-dom.js库
// https://cdnjs.cloudflare.com/ajax/libs/rxjs-dom/7.0.3/rx.dom.compat.min.js

Rx.DOM.get('https://randomuser.me/api/')
  .subscribe(
    function onNext(data) {console.log(data.response)},
    function onError(err) {console.log('Error: ' + err)},   
  )

在Rxjs编程中，可以把所有的数据都当作可观察者对象，不仅是异步数据，比如数组等

常见的操作符

1.将数组变为可观察者对象: from(arr)

Rx.Observable
  .from(['a', 'b', 'c'])
  .subscribe(
    function(x) {console.log('Next: ' + x);}, // 可以不用写方法名'onNext'
    function(err) {console.log('Error: ' + err);},
    function() {console.log('完成');}
  )

//结果
Next: a
Next: b
Next: c
完成

2.将事件变为可观察者对象: fromEvent(dom, eventType)

var allMoves = Rx.Observable
                .fromEvent(document, 'mousemove')

// 订阅
allMoves.subscribe(
  function(e) {console.log(e.clientX, e.clientY);}
)

// 另外还可以对可观察者对象进行转换过滤等
// 在allMoves基础上
var movesOnTheLeft = allMoves.filter(function(e) 
      {return e.clientX < window.innerWidth/2}
    )
movesOnTheLeft.subscribe(function(e) {console.log('鼠标在左边',e.clientX)})

3.将回调函数变为可观察者对象: fromCallback() | fromNodeCallback()

比如Nodejs中

var Rx = require('rx');
var fs = require('fs');

// 从readdir方法中创建可观察者对象
var readdir = Rx.Observable.fromNodeCallback(fs.readdir);

var source = readdir('/users/someone');

var subscription = source.subscribe(
    function(res) {console.log('目录: ' + res);}, // 可以不用写方法名'onNext'
    function(err) {console.log('Error: ' + err);},
    function() {console.log('完成');}
  )

总结

初步认识Rxjs,了解什么是可观察对象，观察者，如何使用操作符来创建它们，本文主要使用 RxJS4.10版本,注意和5.0的区别。

使用到了 rxjs-dom 库来完成ajax请求，初步了解的一些操作符:

• create
• subscribe
• from
• fromEvent
• fromNodeCallback

可观察者的2个方法(接受以observer为参数的回调函数):

• observer.onNext()
• observer.onCompleted()

观察者的3个函数:

• function onNext() {}
• function onError() {}
• function onCompleted() {}