摘要：到底是什么先上代碼輸出這里可以把想象成一個(gè)函數(shù)，這意味著你每次調(diào)用都會(huì)導(dǎo)致傳入里的回調(diào)函數(shù)重新執(zhí)行一次調(diào)用的方式為相當(dāng)于。接收函數(shù)返回值的方式也從改為通過傳入回調(diào)函數(shù)的方式獲取。具體看代碼運(yùn)行結(jié)果如上的第一個(gè)回調(diào)函數(shù)里的結(jié)構(gòu)是推薦的結(jié)構(gòu)。

今早看民工叔的文章的時(shí)候， 發(fā)現(xiàn)對(duì)Rxjs所知甚少， 于是去官方看了下教程， 整理出一些東西， 寫成此文。
Rxjs據(jù)說會(huì)在2017年流行起來， 因?yàn)槠涮幚懋惒竭壿嫞瑪?shù)據(jù)流， 事件非常擅長(zhǎng)。 但是其學(xué)習(xí)曲線相比Promise， EventEmitter陡峭了不少。 而且民工叔也說:"由于RxJS的抽象程度很高，所以，可以用很簡(jiǎn)短代碼表達(dá)很復(fù)雜的含義，這對(duì)開發(fā)人員的要求也會(huì)比較高，需要有比較強(qiáng)的歸納能力。" 本文就Rx.js的幾個(gè)核心概念做出闡述。 盡可能以通俗易懂的方式解釋這些概念。要是本文有誤或不完善的地方，歡迎指出。

先上代碼:

let foo = Rx.Observable.create(observer => {
  console.log("Hello");
  observer.next(42);
});

foo.subscribe(x => console.log(x));
foo.subscribe(y => console.log(y));

輸出

"Hello"
42
"Hello"
42

這里可以把foo想象成一個(gè)函數(shù)，這意味著你每次調(diào)用foo都會(huì)導(dǎo)致傳入Rx.Observable.create里的回調(diào)函數(shù)重新執(zhí)行一次, 調(diào)用的方式為foo.subscribe(callback), 相當(dāng)于foo()。 接收函數(shù)返回值的方式也從var a = foo()改為通過傳入回調(diào)函數(shù)的方式獲取。第三行的observer.next表示返回一個(gè)值, 你可以調(diào)用多次，每次調(diào)用observer.next后， 會(huì)先將next里的值返回給foo.subcribe里的回調(diào)函數(shù), 執(zhí)行完后再返回。observer.complete, observer.error來控制流程。 具體看代碼:

var observable = Rx.Observable.create(observer => {
  try {
    observer.next(1);
    console.log("hello");
    observer.next(2);
    observer.next(3);
    observer.complete();
    observer.next(4);
  } catch (err) {
    observer.error(err); 
  }
});

let = subcription = observable.subscribe(value => {
  console.log(value)
})

運(yùn)行結(jié)果：

1
hello
2
3

如上的第一個(gè)回調(diào)函數(shù)里的結(jié)構(gòu)是推薦的結(jié)構(gòu)。 當(dāng)observable的執(zhí)行出現(xiàn)異常的時(shí)候，通過observer.error將錯(cuò)誤返回, 然而observable.subscribe的回調(diào)函數(shù)無法接收到.因?yàn)閛bserver.complete已經(jīng)調(diào)用, 因此observer.next(4)的返回是無效的. Observable不是可以返回多個(gè)值的Promise 雖然獲得Promise的值的方式也是通過then函數(shù)這種類似的方式, 但是new Promise(callback)里的callback回調(diào)永遠(yuǎn)只會(huì)執(zhí)行一次！因?yàn)?strong>Promise的狀態(tài)是不可逆的。

可以使用其他方式創(chuàng)建Observable， 看代碼：

var clicks = Rx.Observable.fromEvent(document, "click");
clicks.subscribe(x => console.log(x));

當(dāng)用戶對(duì)document產(chǎn)生一個(gè)click行為的時(shí)候， 就會(huì)打印事件對(duì)象到控制臺(tái)上。

先看代碼：

let foo = Rx.Observable.create(observer => {
  console.log("Hello");
  observer.next(42);
});

let observer = x => console.log(x);
foo.subscribe(observer);

代碼中的第二個(gè)變量就是observer. 沒錯(cuò)， observer就是當(dāng)Observable"返回"值的時(shí)候接受那個(gè)值的函數(shù)!第一行中的observer其實(shí)就是通過foo.subscribe傳入的callback. 只不過稍加封裝了。 怎么封裝的？ 看代碼：

let foo = Rx.Observable.create(observer => {
  try {
    console.log("Hello");
    observer.next(42);
    observer.complete();
    observer.next(10);
  } catch(e) { observer.error(e) }
  
});

let observer = {
  next(value) { console.log(value) },
  complete() { console.log("completed"),
  error(err) { console.error(err) }
}
foo.subscribe(observer);

你看到observer被定義成了一個(gè)對(duì)象， 其實(shí)這才是完整的observer. 傳入一個(gè)callback到observable.subcribe相當(dāng)于傳入了{ next: callback }。

Subscription是什么， 先上代碼：

var observable = Rx.Observable.interval(1000);
var subscription = observable.subscribe(x => console.log(x));

setTimeout(() => {
  subscription.unsubscribe();
}, 3100)

運(yùn)行結(jié)果：

0
1
2

Rx.Observable.interval可以返回一個(gè)能夠發(fā)射(返回)0， 1， 2， 3...， n數(shù)字的Observable， 返回的時(shí)間間隔這里是1000ms。 第二行中的變量就是subscription。 subscription有一個(gè)unsubscribe方法, 這個(gè)方法可以讓subscription訂閱的observable發(fā)射的數(shù)據(jù)被observer忽略掉.通俗點(diǎn)說就是取消訂閱。

unsubscribe存在一個(gè)陷阱。 先看代碼：

var foo = Rx.Observable.create((observer) => {
  var i = 0
  setInterval(() => {
    observer.next(i++)
    console.log("hello")
  }, 1000)
})

const subcription = foo.subscribe((i) => console.log(i))
subcription.unsubscribe()

運(yùn)行結(jié)果：

hello
hello
hello
......
hello

unsubscribe只會(huì)讓observer忽略掉observable發(fā)射的數(shù)據(jù)，但是setInterval依然會(huì)繼續(xù)執(zhí)行。 這看起來似乎是一個(gè)愚蠢的設(shè)計(jì)。 所以不建議這樣寫。

Subject是一種能夠發(fā)射數(shù)據(jù)給多個(gè)observer的Observable, 這讓Subject看起來就好像是EventEmitter。 先上代碼：

var subject = new Rx.Subject();

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});
subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(1);
subject.next(2);

運(yùn)行結(jié)果：

observerA: 1
observerB: 1
observerA: 2
observerB: 2

與Observable不同的是， Subject發(fā)射數(shù)據(jù)給多個(gè)observer。 其次， 定義subject的時(shí)候并沒有傳入callback， 這是因?yàn)閟ubject自帶next, complete, error等方法。從而可以發(fā)射數(shù)據(jù)給observer。 這和EventEmitter很類似。observer并不知道他subscribe的是Obervable還是Subject。 對(duì)observer來說是透明的。 而且Subject還有各種派生， 比如說：

BehaviorSubject 能夠保留最近的數(shù)據(jù)，使得當(dāng)有subscribe的時(shí)候，立馬發(fā)射出去。看代碼：

var subject = new Rx.BehaviorSubject(0); // 0 is the initial value

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);

subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(3);

運(yùn)行結(jié)果：

observerA: 0
observerA: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3

ReplaySubject 能夠保留最近的一些數(shù)據(jù)， 使得當(dāng)有subscribe的時(shí)候，將這些數(shù)據(jù)發(fā)射出去。看代碼：

var subject = new Rx.ReplaySubject(3); 

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(5);

輸出結(jié)果：

observerA: 1
observerA: 2
observerA: 3
observerA: 4
observerB: 2
observerB: 3
observerB: 4
observerA: 5
observerB: 5

第一行的聲明表示ReplaySubject最大能夠記錄的數(shù)據(jù)的數(shù)量是3。

AsyncSubject 只會(huì)發(fā)射結(jié)束前的一個(gè)數(shù)據(jù)。 看代碼：

var subject = new Rx.AsyncSubject();

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(5);
subject.complete();

輸出結(jié)果：

observerA: 5
observerB: 5

既然subject有next, error, complete三種方法， 那subject就可以作為observer！ 看代碼：

var subject = new Rx.Subject();

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});
subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

var observable = Rx.Observable.from([1, 2, 3]);

observable.subscribe(subject);

輸出結(jié)果：

observerA: 1
observerB: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3

也就是說， observable.subscribe可以傳入一個(gè)subject來訂閱其消息。 這就好像是Rxjs中的一顆語法糖， Rxjs有專門的實(shí)現(xiàn)。

Multicasted Observables 是一種借助Subject來將數(shù)據(jù)發(fā)射給多個(gè)observer的Observable。 看代碼：

var source = Rx.Observable.from([1, 2, 3]);
var subject = new Rx.Subject();
var multicasted = source.multicast(subject);

multicasted.subscribe({
  next: (v) => console.log("observerA: " + v)
});
multicasted.subscribe({
  next: (v) => console.log("observerB: " + v)
});

multicasted.connect();

Rx.Observable.from能夠逐一發(fā)射數(shù)組中的元素， 在multicasted.connect()調(diào)用之前的任何subscribe都不會(huì)導(dǎo)致source發(fā)射數(shù)據(jù)。multicasted.connect()相當(dāng)于之前的observable.subscribe(subject)。因此不能將multicasted.connect()寫在subscribe的前面。因?yàn)檫@會(huì)導(dǎo)致在執(zhí)行multicasted.connect()的時(shí)候source發(fā)射數(shù)據(jù)， 但是subject又沒保存數(shù)據(jù)， 導(dǎo)致兩個(gè)subscribe無法接收到任何數(shù)據(jù)。

最好是第一個(gè)subscribe的時(shí)候能夠得到當(dāng)前已有的數(shù)據(jù)， 最后一個(gè)unsubscribe的時(shí)候就停止Observable的執(zhí)行， 相當(dāng)于Observable發(fā)射的數(shù)據(jù)都被忽略。

refCount就是能夠返回這樣的Observable的方法

var source = Rx.Observable.interval(500);
var subject = new Rx.Subject();
var refCounted = source.multicast(subject).refCount();
var subscription1, subscription2, subscriptionConnect;

console.log("observerA subscribed");
subscription1 = refCounted.subscribe({
  next: (v) => console.log("observerA: " + v)
});

setTimeout(() => {
  console.log("observerB subscribed");
  subscription2 = refCounted.subscribe({
    next: (v) => console.log("observerB: " + v)
  });
}, 600);

setTimeout(() => {
  console.log("observerA unsubscribed");
  subscription1.unsubscribe();
}, 1200);

setTimeout(() => {
  console.log("observerB unsubscribed");
  subscription2.unsubscribe();
}, 2000);

輸出結(jié)果：

observerA subscribed
observerA: 0
observerB subscribed
observerA: 1
observerB: 1
observerA unsubscribed
observerB: 2
observerB unsubscribed

Observable上有很多方法， 比如說map, filter, merge等等。 他們基于調(diào)用它們的observable，返回一個(gè)全新的observable。 而且他們都是純方法。 operators分為兩種， instance operators 和 static operators。 instance operators是存在于observable實(shí)例上的方法， 也就是實(shí)例方法； static operators是存在于Observable這個(gè)類型上的方法， 也就是靜態(tài)方法。Rxjs擁有很多強(qiáng)大的operators。

自己實(shí)現(xiàn)一個(gè)operators：

function multiplyByTen(input) {
  var output = Rx.Observable.create(function subscribe(observer) {
    input.subscribe({
      next: (v) => observer.next(10 * v),
      error: (err) => observer.error(err),
      complete: () => observer.complete()
    });
  });
  return output;
}

var input = Rx.Observable.from([1, 2, 3, 4]);
var output = multiplyByTen(input);
output.subscribe(x => console.log(x));

輸出結(jié)果：

10
20
30
40

import React from "react";
import ReactDOM from "react-dom";
import Rx from "rx";

class Main extends React.Component {
  constructor (props) {
    super(props);
    this.state = {count: 0};
  }

  // Click events are now observables! No more proactive approach.
  componentDidMount () {
    const plusBtn = document.getElementById("plus");
    const minusBtn = document.getElementById("minus");

    const plus$ = Rx.Observable.fromEvent(plusBtn, "click").map(e => 1);
    const minus$ = Rx.Observable.fromEvent(minusBtn, "click").map(e => -1);

    Rx.Observable.merge(plus$, minus$).scan((acc, n) => acc + n)
      .subscribe(value => this.setState({count: value}));
  }

  render () {
    return (
        

          +
          -
          
count: {this.state.count}
        
    );
  }
}

ReactDOM.render(
, document.getElementById("app"));

merge用于合并兩個(gè)observable產(chǎn)生一個(gè)新的observable。 scan類似于Array中的reduce。 這個(gè)例子實(shí)現(xiàn)了點(diǎn)擊plus的時(shí)候+1， 點(diǎn)擊minus的時(shí)候-1。

多個(gè)復(fù)雜的異步或事件組合在一起。

處理多個(gè)數(shù)據(jù)序列

假如沒有被復(fù)雜的異步，事件， 數(shù)據(jù)序列困擾， 如果promise已經(jīng)足夠的話， 就沒必要適用Rx.js。

Observable， Observer， Subscription， Subscrib， Subject概念。

RxJS適用于解決復(fù)雜的異步，事件問題。

讓我們一起來學(xué)習(xí) RxJS ---by 餓了么前端

21-use-rxjs-for-orchestrating-asynchronous-and-event-based-computations

RxJS文檔

RxJS 入門指引和初步應(yīng)用 ---by 民工叔