Angular9 RxJS庫

響應(yīng)式編程是一種面向數(shù)據(jù)流和變更傳播的異步編程范式（Wikipedia）。RxJS（響應(yīng)式擴展的 JavaScript 版）是一個使用可觀察對象進(jìn)行響應(yīng)式編程的庫，它讓組合異步代碼和基于回調(diào)的代碼變得更簡單。參見 RxJS 官方文檔。

RxJS 提供了一種對 Observable 類型的實現(xiàn)，直到 Observable 成為了 JavaScript 語言的一部分并且瀏覽器支持它之前，它都是必要的。這個庫還提供了一些工具函數(shù)，用于創(chuàng)建和使用可觀察對象。這些工具函數(shù)可用于：

• 把現(xiàn)有的異步代碼轉(zhuǎn)換成可觀察對象

• 迭代流中的各個值

• 把這些值映射成其它類型

• 對流進(jìn)行過濾

• 組合多個流

創(chuàng)建可觀察對象的函數(shù)

RxJS 提供了一些用來創(chuàng)建可觀察對象的函數(shù)。這些函數(shù)可以簡化根據(jù)某些東西創(chuàng)建可觀察對象的過程，比如事件、定時器、承諾等等。比如：

//Create an observable from a promise


import { from } from 'rxjs';


// Create an Observable out of a promise
const data = from(fetch('/api/endpoint'));
// Subscribe to begin listening for async result
data.subscribe({
  next(response) { console.log(response); },
  error(err) { console.error('Error: ' + err); },
  complete() { console.log('Completed'); }
});

//Create an observable from a counter


import { interval } from 'rxjs';


// Create an Observable that will publish a value on an interval
const secondsCounter = interval(1000);
// Subscribe to begin publishing values
secondsCounter.subscribe(n =>
  console.log(`It's been ${n} seconds since subscribing!`));

//Create an observable from an event


import { fromEvent } from 'rxjs';


const el = document.getElementById('my-element');


// Create an Observable that will publish mouse movements
const mouseMoves = fromEvent(el, 'mousemove');


// Subscribe to start listening for mouse-move events
const subscription = mouseMoves.subscribe((evt: MouseEvent) => {
  // Log coords of mouse movements
  console.log(`Coords: ${evt.clientX} X ${evt.clientY}`);


  // When the mouse is over the upper-left of the screen,
  // unsubscribe to stop listening for mouse movements
  if (evt.clientX < 40 && evt.clientY < 40) {
    subscription.unsubscribe();
  }
});

//Create an observable that creates an AJAX request


import { ajax } from 'rxjs/ajax';


// Create an Observable that will create an AJAX request
const apiData = ajax('/api/data');
// Subscribe to create the request
apiData.subscribe(res => console.log(res.status, res.response));

操作符

操作符是基于可觀察對象構(gòu)建的一些對集合進(jìn)行復(fù)雜操作的函數(shù)。RxJS 定義了一些操作符，比如 map()、filter()、concat() 和 flatMap()。

操作符接受一些配置項，然后返回一個以來源可觀察對象為參數(shù)的函數(shù)。當(dāng)執(zhí)行這個返回的函數(shù)時，這個操作符會觀察來源可觀察對象中發(fā)出的值，轉(zhuǎn)換它們，并返回由轉(zhuǎn)換后的值組成的新的可觀察對象。下面是一個簡單的例子：

Path："Map operator" 。

import { map } from 'rxjs/operators';


const nums = of(1, 2, 3);


const squareValues = map((val: number) => val * val);
const squaredNums = squareValues(nums);


squaredNums.subscribe(x => console.log(x));


// Logs
// 1
// 4
// 9

你可以使用管道來把這些操作符鏈接起來。管道讓你可以把多個由操作符返回的函數(shù)組合成一個。pipe() 函數(shù)以你要組合的這些函數(shù)作為參數(shù)，并且返回一個新的函數(shù)，當(dāng)執(zhí)行這個新函數(shù)時，就會順序執(zhí)行那些被組合進(jìn)去的函數(shù)。

應(yīng)用于某個可觀察對象上的一組操作符就像一個處理流程 —— 也就是說，對你感興趣的這些值進(jìn)行處理的一組操作步驟。這個處理流程本身不會做任何事。你需要調(diào)用 subscribe() 來通過處理流程得出并生成一個結(jié)果。

例子如下：

Path："Standalone pipe function" 。

import { filter, map } from 'rxjs/operators';


const nums = of(1, 2, 3, 4, 5);


// Create a function that accepts an Observable.
const squareOddVals = pipe(
  filter((n: number) => n % 2 !== 0),
  map(n => n * n)
);


// Create an Observable that will run the filter and map functions
const squareOdd = squareOddVals(nums);


// Subscribe to run the combined functions
squareOdd.subscribe(x => console.log(x));

pipe() 函數(shù)也同時是 RxJS 的 Observable 上的一個方法，所以你可以用下列簡寫形式來達(dá)到同樣的效果：

Path："Observable.pipe function" 。

import { filter, map } from 'rxjs/operators';


const squareOdd = of(1, 2, 3, 4, 5)
  .pipe(
    filter(n => n % 2 !== 0),
    map(n => n * n)
  );


// Subscribe to get values
squareOdd.subscribe(x => console.log(x));

常用操作符

RxJS 提供了很多操作符，不過只有少數(shù)是常用的。 下面是一個常用操作符的列表和用法范例，參見 [RxJS API]() 文檔。

注：

• 對于 Angular 應(yīng)用來說，我們提倡使用管道來組合操作符，而不是使用鏈?zhǔn)綄懛?。鏈?zhǔn)綄懛ㄈ匀辉诤芏?RxJS 中使用著。

錯誤處理

除了可以在訂閱時提供 error() 處理器外，RxJS 還提供了 catchError 操作符，它允許你在管道中處理已知錯誤。

假設(shè)你有一個可觀察對象，它發(fā)起 API 請求，然后對服務(wù)器返回的響應(yīng)進(jìn)行映射。如果服務(wù)器返回了錯誤或值不存在，就會生成一個錯誤。如果你捕獲這個錯誤并提供了一個默認(rèn)值，流就會繼續(xù)處理這些值，而不會報錯。

下面是使用 catchError 操作符實現(xiàn)這種效果的例子：

Path："catchError operator" 。

import { ajax } from 'rxjs/ajax';
import { map, catchError } from 'rxjs/operators';
// Return "response" from the API. If an error happens,
// return an empty array.
const apiData = ajax('/api/data').pipe(
  map(res => {
    if (!res.response) {
      throw new Error('Value expected!');
    }
    return res.response;
  }),
  catchError(err => of([]))
);


apiData.subscribe({
  next(x) { console.log('data: ', x); },
  error(err) { console.log('errors already caught... will not run'); }
});

重試失敗的可觀察對象

catchError 提供了一種簡單的方式進(jìn)行恢復(fù)，而 retry 操作符讓你可以嘗試失敗的請求。

可以在 catchError 之前使用 retry 操作符。它會訂閱到原始的來源可觀察對象，它可以重新運行導(dǎo)致結(jié)果出錯的動作序列。如果其中包含 HTTP 請求，它就會重新發(fā)起那個 HTTP 請求。

下列代碼把前面的例子改成了在捕獲錯誤之前重發(fā)請求：

Path："retry operator" 。

import { ajax } from 'rxjs/ajax';
import { map, retry, catchError } from 'rxjs/operators';


const apiData = ajax('/api/data').pipe(
  retry(3), // Retry up to 3 times before failing
  map(res => {
    if (!res.response) {
      throw new Error('Value expected!');
    }
    return res.response;
  }),
  catchError(err => of([]))
);


apiData.subscribe({
  next(x) { console.log('data: ', x); },
  error(err) { console.log('errors already caught... will not run'); }
});

不要重試登錄認(rèn)證請求，這些請求只應(yīng)該由用戶操作觸發(fā)。我們肯定不會希望自動重復(fù)發(fā)送登錄請求導(dǎo)致用戶的賬號被鎖定。

可觀察對象的命名約定

由于 Angular 的應(yīng)用幾乎都是用 TypeScript 寫的，你通常會希望知道某個變量是否可觀察對象。雖然 Angular 框架并沒有針對可觀察對象的強制性命名約定，不過你經(jīng)常會看到可觀察對象的名字以“$”符號結(jié)尾。

這在快速瀏覽代碼并查找可觀察對象值時會非常有用。同樣的，如果你希望用某個屬性來存儲來自可觀察對象的最近一個值，它的命名慣例是與可觀察對象同名，但不帶“$”后綴。

比如：

Path："Naming observables" 。

import { Component } from '@angular/core';
import { Observable } from 'rxjs';


@Component({
  selector: 'app-stopwatch',
  templateUrl: './stopwatch.component.html'
})
export class StopwatchComponent {


  stopwatchValue: number;
  stopwatchValue$: Observable<number>;


  start() {
    this.stopwatchValue$.subscribe(num =>
      this.stopwatchValue = num
    );
  }
}