摘要：是一種針對異步數據流編程工具，或者叫響應式擴展編程可不管如何解釋其目標就是異步編程，引入為了就是讓異步可控更簡單。最合理的方式在調用它。當組件需要向組件傳遞數據時，我們可以在組件中使用。的作用是使指令或組件能自定義事件。

RxJS是一種針對異步數據流編程工具，或者叫響應式擴展編程；可不管如何解釋RxJS其目標就是異步編程，Angular引入RxJS為了就是讓異步可控、更簡單。

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

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

當Observable設置觀察者后，而連接并獲取原始數據的這個過程叫生產者，可能是DOM中的 click 事件、input 事件、或者更加復雜的HTTP通信。

為了更好理解，先從一個簡單的示例開始：

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]");

    // 第二個參數 input 是事件名，對于input元素有一個 oninput 事件用于接受用戶輸入
    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() 會返回一個Observable，并且監聽 input 事件，當事件被觸發后會發送一個 Event 給對應的observer觀察者。

observer非常簡單，像上面示例中 subscribe 訂閱就是接收一個 observer 方法。

一般在Angular我們 subscribe 會這么寫：

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

});

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

當Observable產生一個新值時，會通知 observer 的 next()，而當捕獲失敗可以調用 error()。

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

Observable的生產的值允許經過一序列格式化或操作，最終得到一個有價值的數據給觀察者，而這一切是由一序列鏈式operator來完成的，每一個operator都會產生一個新的Observable。而我們也稱這一序列過程為：流。

正如前面說到的，Observable可以鏈式寫法，這意味著我們可以這樣：

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

下面是整個順序步驟：

假設用戶輸入：a

Observable對觸發 oninput 事件作出反應，將值以參數的形式傳遞給observer的 next()。

map() 根據 event.target.value 的內容返回一個新的 Observable，并調用 next() 傳遞給下一個observer。

filter() 如果值長度 >=2 的話，則返回一個新的 Observable，并調用 next() 傳遞給下一個observer。

最后，將結果傳遞給 subscribe 訂閱塊。

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

為什么需要取消訂閱

Observable 當有數據產生時才會推送給訂閱者，所以它可能會無限次向訂閱者推送數據。正因為如此，在Angular里面創建組件的時候務必要取消訂閱操作，以避免內存泄漏，要知道在SPA世界里懂得擦屁股是一件必須的事。

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

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

如果組件有很多訂閱者的話，則需要將這些訂閱者存儲在數組中，并組件被銷毀時再逐個取消訂閱。但，我們有更好的辦法：

使用 [takeWhile()
](http://reactivex.io/documenta... operator，它會在你傳遞一個布爾值是調用 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;
}

簡單有效，而且優雅。

如果說 Observable 與 observer 是攻受結合體的話，那么 Subject 就是一個人即攻亦受。正因為如此，我們在寫一個Service用于數據傳遞時，總是使用 new Subject。

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

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

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

當F組件需要向M組件傳遞數據時，我們可以在F組件中使用 send()。

constructor(public srv: MessageService) { }

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

而M組件只需要訂閱內容就行：

constructor(private srv: MessageService) {}

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

其實EventEmitter跟RxJS沒有直接關系，因為他是Angular的產物，而非RxJS的東西。或者我們壓根沒必要去談，因為EventEmitter就是Subject。

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

@Output() changed = new EventEmitter();

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

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

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

RxJS最難我想就是各種operator的應用了，這需要一些經驗的積累。

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

創建數據流：

單值：of, empty, never

多值：from

定時：interval, timer

從事件創建：fromEvent

從Promise創建：fromPromise

自定義創建：create

轉換操作：

改變數據形態：map, mapTo, pluck

過濾一些值：filter, skip, first, last, take

時間軸上的操作：delay, timeout, throttle, debounce, audit, bufferTime

累加：reduce, scan

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

條件執行：takeUntil, delayWhen, retryWhen, subscribeOn, ObserveOn

轉接：switch

組合數據流：

concat，保持原來的序列順序連接兩個數據流

merge，合并序列

race，預設條件為其中一個數據流完成

forkJoin，預設條件為所有數據流都完成

zip，取各來源數據流最后一個值合并為對象

combineLatest，取各來源數據流最后一個值合并為數組

另，最好使用 **$** 結尾的命名方式來表示Observable，例：input$。

happy coding!