摘要：是一種針對(duì)異步數(shù)據(jù)流編程工具，或者叫響應(yīng)式擴(kuò)展編程可不管如何解釋其目標(biāo)就是異步編程，引入為了就是讓異步可控更簡(jiǎn)單。最合理的方式在調(diào)用它。當(dāng)組件需要向組件傳遞數(shù)據(jù)時(shí)，我們可以在組件中使用。的作用是使指令或組件能自定義事件。

RxJS是一種針對(duì)異步數(shù)據(jù)流編程工具，或者叫響應(yīng)式擴(kuò)展編程；可不管如何解釋RxJS其目標(biāo)就是異步編程，Angular引入RxJS為了就是讓異步可控、更簡(jiǎn)單。

而今就是要探討什么是Observable、observer、operator、Submit、EventEmmit，以及如何去使用它們。

Observable只是一個(gè)普通函數(shù)，要想讓他有所作為，就需要跟observer一起使用；前者是受后者是攻。而這個(gè)observer（后面我們會(huì)介紹）只是一個(gè)帶有 next、error、complete 的簡(jiǎn)單對(duì)象而已。最后，還需要通過(guò) subscribe 訂閱來(lái)啟動(dòng)Observable；否則它是不會(huì)有任何反應(yīng)；可以理解為陌*為了他們能在一起而提供的環(huán)境，而訂閱也會(huì)返回一個(gè)可用于取消操作（在RxJS里叫 unsubscribe）。

當(dāng)Observable設(shè)置觀察者后，而連接并獲取原始數(shù)據(jù)的這個(gè)過(guò)程叫生產(chǎn)者，可能是DOM中的 click 事件、input 事件、或者更加復(fù)雜的HTTP通信。

為了更好理解，先從一個(gè)簡(jiǎn)單的示例開(kāi)始：

import { Component } from "@angular/core";
import { Observable, Subscription } from "rxjs";

@Component({
  selector: "app-home",
  template: ` `
})
export class HomeComponent {
  ngOnInit() {
    const node = document.querySelector("input[type=text]");

    // 第二個(gè)參數(shù) input 是事件名，對(duì)于input元素有一個(gè) oninput 事件用于接受用戶(hù)輸入
    const input$ = Observable.fromEvent(node, "input");
    input$.subscribe({
      next: (event: any) => console.log(`You just typed ${event.target.value}!`),
      error: (err) => console.log(`Oops... ${err}`),
      complete: () => console.log(`Complete!`)
    });
  }
}

示例中 Observable.fromEvent() 會(huì)返回一個(gè)Observable，并且監(jiān)聽(tīng) input 事件，當(dāng)事件被觸發(fā)后會(huì)發(fā)送一個(gè) Event 給對(duì)應(yīng)的observer觀察者。

observer非常簡(jiǎn)單，像上面示例中 subscribe 訂閱就是接收一個(gè) observer 方法。

一般在Angular我們 subscribe 會(huì)這么寫(xiě)：

input$.subscribe((event: any) => {

});

從語(yǔ)法角度來(lái)講和 subscribe({ next, error, complete }) 是一樣的。

當(dāng)Observable產(chǎn)生一個(gè)新值時(shí)，會(huì)通知 observer 的 next()，而當(dāng)捕獲失敗可以調(diào)用 error()。

當(dāng)Observable被訂閱后，除非調(diào)用observer的 complete() 或 unsubscribe() 取消訂閱兩情況以外；會(huì)一直將值傳遞給 observer。

Observable的生產(chǎn)的值允許經(jīng)過(guò)一序列格式化或操作，最終得到一個(gè)有價(jià)值的數(shù)據(jù)給觀察者，而這一切是由一序列鏈?zhǔn)給perator來(lái)完成的，每一個(gè)operator都會(huì)產(chǎn)生一個(gè)新的Observable。而我們也稱(chēng)這一序列過(guò)程為：流。

正如前面說(shuō)到的，Observable可以鏈?zhǔn)綄?xiě)法，這意味著我們可以這樣：

Observable.fromEvent(node, "input")
  .map((event: any) => event.target.value)
  .filter(value => value.length >= 2)
  .subscribe(value => { console.log(value); });

下面是整個(gè)順序步驟：

假設(shè)用戶(hù)輸入：a

Observable對(duì)觸發(fā) oninput 事件作出反應(yīng)，將值以參數(shù)的形式傳遞給observer的 next()。

map() 根據(jù) event.target.value 的內(nèi)容返回一個(gè)新的 Observable，并調(diào)用 next() 傳遞給下一個(gè)observer。

filter() 如果值長(zhǎng)度 >=2 的話，則返回一個(gè)新的 Observable，并調(diào)用 next() 傳遞給下一個(gè)observer。

最后，將結(jié)果傳遞給 subscribe 訂閱塊。

你只要記住每一次 operator 都會(huì)返回一個(gè)新的 Observable，不管 operator 有多少個(gè)，最終只有最后一個(gè) Observable 會(huì)被訂閱。

為什么需要取消訂閱

Observable 當(dāng)有數(shù)據(jù)產(chǎn)生時(shí)才會(huì)推送給訂閱者，所以它可能會(huì)無(wú)限次向訂閱者推送數(shù)據(jù)。正因?yàn)槿绱?，在Angular里面創(chuàng)建組件的時(shí)候務(wù)必要取消訂閱操作，以避免內(nèi)存泄漏，要知道在SPA世界里懂得擦屁股是一件必須的事。

前面示例講過(guò)，調(diào)用 subscribe() 后，會(huì)返回一個(gè) Subscription 可用于取消操作 unsubscribe()。最合理的方式在 ngOnDestroy 調(diào)用它。

ngOnDestroy() {
    this.inputSubscription.unsubscribe();
}

如果組件有很多訂閱者的話，則需要將這些訂閱者存儲(chǔ)在數(shù)組中，并組件被銷(xiāo)毀時(shí)再逐個(gè)取消訂閱。但，我們有更好的辦法：

使用 [takeWhile()
](http://reactivex.io/documenta... operator，它會(huì)在你傳遞一個(gè)布爾值是調(diào)用 next() 還是 complete()。

private alive: boolean = true;
ngOnInit() {
  const node = document.querySelector("input[type=text]");

  this.s = Observable.fromEvent(node, "input")
    .takeWhile(() => this.alive)
    .map((event: any) => event.target.value)
    .filter(value => value.length >= 2)
    .subscribe(value => { console.log(value) });
}

ngOnDestroy() {
  this.alive = false;
}

簡(jiǎn)單有效，而且優(yōu)雅。

如果說(shuō) Observable 與 observer 是攻受結(jié)合體的話，那么 Subject 就是一個(gè)人即攻亦受。正因?yàn)槿绱?，我們?cè)趯?xiě)一個(gè)Service用于數(shù)據(jù)傳遞時(shí)，總是使用 new Subject。

@Injectable()
export class MessageService {
    private subject = new Subject();

    send(message: any) {
        this.subject.next(message);
    }

    get(): Observable {
        return this.subject.asObservable();
    }
}

當(dāng)F組件需要向M組件傳遞數(shù)據(jù)時(shí)，我們可以在F組件中使用 send()。

constructor(public srv: MessageService) { }

ngOnInit() {
    this.srv.send("w s k f m?")
}

而M組件只需要訂閱內(nèi)容就行：

constructor(private srv: MessageService) {}

message: any;
ngOnInit() {
    this.srv.get().subscribe((result) => {
        this.message = result;
    })
}

其實(shí)EventEmitter跟RxJS沒(méi)有直接關(guān)系，因?yàn)樗茿ngular的產(chǎn)物，而非RxJS的東西?；蛘呶覀儔焊鶝](méi)必要去談，因?yàn)镋ventEmitter就是Subject。

EventEmitter的作用是使指令或組件能自定義事件。

@Output() changed = new EventEmitter();

click() {
    this.changed.emit("hi~");
}

@Component({
  template: ``
})
export class HomeComponent {
  subscribe(message: string) {
     // 接收：hi~
  }
}

上面示例其實(shí)和上一個(gè)示例中 MessageService 如出一轍，只不過(guò)是將 next() 換成 emit() 僅此而已。

RxJS最難我想就是各種operator的應(yīng)用了，這需要一些經(jīng)驗(yàn)的積累。

RxJS很火很大原因我認(rèn)還是提供了豐富的API，以下是摘抄：

創(chuàng)建數(shù)據(jù)流：

單值：of, empty, never

多值：from

定時(shí)：interval, timer

從事件創(chuàng)建：fromEvent

從Promise創(chuàng)建：fromPromise

自定義創(chuàng)建：create

轉(zhuǎn)換操作：

改變數(shù)據(jù)形態(tài)：map, mapTo, pluck

過(guò)濾一些值：filter, skip, first, last, take

時(shí)間軸上的操作：delay, timeout, throttle, debounce, audit, bufferTime

累加：reduce, scan

異常處理：throw, catch, retry, finally

條件執(zhí)行：takeUntil, delayWhen, retryWhen, subscribeOn, ObserveOn

轉(zhuǎn)接：switch

組合數(shù)據(jù)流：

concat，保持原來(lái)的序列順序連接兩個(gè)數(shù)據(jù)流

merge，合并序列

race，預(yù)設(shè)條件為其中一個(gè)數(shù)據(jù)流完成

forkJoin，預(yù)設(shè)條件為所有數(shù)據(jù)流都完成

zip，取各來(lái)源數(shù)據(jù)流最后一個(gè)值合并為對(duì)象

combineLatest，取各來(lái)源數(shù)據(jù)流最后一個(gè)值合并為數(shù)組

另，最好使用 **$** 結(jié)尾的命名方式來(lái)表示Observable，例：input$。

happy coding!