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

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

什么是Observable？

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: `<input type="text"> `})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？

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。而我們也稱這一序列過程為：流。

什么是operator？

正如前面說到的，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世界里懂得擦屁股是一件必須的事。

unsubscribe

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

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

takeWhile

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

使用 takeWhile() 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;}

簡單有效，而且優雅。

Subject

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

@Injectable()export class MessageService {  private subject = new Subject<any>();  send(message: any) {    this.subject.next(message);  }  get(): Observable<any> {    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

其實EventEmitter跟RxJS沒有直接關系，因為他是Angular的產物，而非RxJS的東西?；蛘呶覀儔焊鶝]必要去談，因為EventEmitter就是Subject。

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

@Output() changed = new EventEmitter<string>();click() {  this.changed.emit('hi~');}

@Component({ template: `<comp (changed)="subscribe($event)"></comp>`})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$。

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持武林網。