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

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

什么是Observable？

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)單的示例開始：

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

  // 第二個(gè)參數(shù) input 是事件名，對(duì)于input元素有一個(gè) 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() 會(huì)返回一個(gè)Observable，并且監(jiān)聽 input 事件，當(dāng)事件被觸發(fā)后會(huì)發(fā)送一個(gè) Event 給對(duì)應(yīng)的observer觀察者。

什么是observer？

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

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

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

什么是operator？

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

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

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

• 假設(shè)用戶輸入：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世界里懂得擦屁股是一件必須的事。

unsubscribe

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

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

takeWhile

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

使用 takeWhile() 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)雅。

Subject

如果說(shuō) Observable 與 observer 是攻受結(jié)合體的話，那么 Subject 就是一個(gè)人即攻亦受。正因?yàn)槿绱?，我們?cè)趯懸粋€(gè)Service用于數(shù)據(jù)傳遞時(shí)，總是使用 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();
  }
}

當(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;
  })
}

EventEmitter

其實(shí)EventEmitter跟RxJS沒有直接關(guān)系，因?yàn)樗茿ngular的產(chǎn)物，而非RxJS的東西。或者我們壓根沒必要去談，因?yàn)镋ventEmitter就是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~
 }
}

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

結(jié)論

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$。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論