淺析Angular中的可觀察對象、觀察者和RxJS操作符

本篇文章帶大家了解Angular中的可觀察對象（Observable）、觀察者（observer）和RxJS操作符，希望對大家有所幫助！

Observable（可觀察對象）

Observable(可觀察對象)，是RxJS庫里面的一個對象，可以用來處理異步事件，例如HTTP請求（實際上，在Angular中，所有的HTTP請求返回的都是Observable）。【相關教程推薦：《angular教程》】

或許，你以前接觸過一個叫promise的東西，它們本質上面是相同的：都是生產者主動向消費者“push”產品，而消費者是被動接收的，但是他們兩者還是有很大區(qū)別的：Observable可以發(fā)送任意多值，并且，在被訂閱之前，它是不會執(zhí)行的！這是promise不具備的特點。

• Observable用于在發(fā)送方和接收方之間傳輸消息，你可以將這些消息看成是流
• 在創(chuàng)建Observable對象時，需要傳入一個函數(shù)作為構造函數(shù)的參數(shù)，這個函數(shù)叫訂閱者函數(shù)，這個函數(shù)也就是生產者向消費者推送消息的地方
• 在被消費者subscribe（訂閱）之前，訂閱者函數(shù)不會被執(zhí)行，直到subscribe()函數(shù)被調用，該函數(shù)返回一個subscription對象，里面有一個unsubscribe()函數(shù)，消費者可以隨時拒絕消息的接收！
• subscribe()函數(shù)接收一個observer（觀察者）對象作為入?yún)?/li>
• 消息的發(fā)送可以是同步的，也可以是異步的

observer（觀察者）

有了可觀察對象（發(fā)送方），就需要一個觀察者（接收方）來觀察可觀察對象，觀察者要實現(xiàn)observer接口，它是一個對象，其中包含三個屬性，它們都是函數(shù)，如下:

訂閱

只有當有人訂閱 Observable 的實例時，它才會開始發(fā)布值。 訂閱時要先調用可觀察對象的 subscribe() 方法，并把一個觀察者對象傳給它，用來接收通知。如下：

為了展示訂閱的原理，需要先創(chuàng)建新的可觀察對象。它有一個構造函數(shù)可以用來創(chuàng)建新實例，但是為了更簡明，也可以使用 Observable 上定義的一些靜態(tài)方法來創(chuàng)建一些常用的簡單可觀察對象：

• of(...items) ：返回一個 Observable 實例，它用同步的方式把參數(shù)中提供的這些值一個一個發(fā)送出來。
• from(iterable) ： 把它的參數(shù)轉換成一個 Observable 實例。 該方法通常用于把一個數(shù)組轉換成一個（發(fā)送多個值的）可觀察對象。

import { of } from "rxjs"; // 1、通過 of() 方法返回一個可觀察對象，并準備將1，2，3三個數(shù)據(jù)發(fā)送出去 const observable = of(1, 2, 3);	 // 2、實現(xiàn) observer 接口，觀察者 const observer = {	 	next: (num: number) => console.log(num), 	error: (err: Error) => console.error('Observer got an error: ' + err),   	complete: () => console.log('Observer got a complete notification'),  } // 3、訂閱。調用可觀察對象的 subscribe() 方法訂閱，subscribe() 方法中傳入的對象就是一個觀察者 observable.subscribe(observer);

運行結果如下：

上面訂閱的寫法可以直接改為如下：參數(shù)不是對象

observable.subscribe(   num => console.log(num),   err => console.error('Observer got an error: ' + err),   () => console.log('Observer got a complete notification') );

訂閱者函數(shù)

在上面的例子中使用的是of()方法來創(chuàng)建可觀察對象，這節(jié)使用構造函數(shù)創(chuàng)建可觀察對象。

Observable 構造函數(shù)可以創(chuàng)建任何類型的可觀察流。 當執(zhí)行可觀察對象的 subscribe() 方法時，這個構造函數(shù)就會把它接收到的參數(shù)作為訂閱函數(shù)來運行。 訂閱函數(shù)會接收一個 Observer 對象，并把值發(fā)布給觀察者的 next()方法。

// 1、自定義訂閱者函數(shù) function sequenceSubscriber(observer: Observer<number>) {   observer.next(1);	// 發(fā)送數(shù)據(jù)   observer.next(2);	// 發(fā)送數(shù)據(jù)   observer.next(3);	// 發(fā)送數(shù)據(jù)   observer.complete();   return {unsubscribe() {}}; }  // 2、通過構造函數(shù)創(chuàng)建一個新的可觀察對象，參數(shù)就是一個訂閱者函數(shù) const sequence = new Observable(sequenceSubscriber);  // 3、訂閱 sequence.subscribe({   next(num) { console.log(num); },	// 接受數(shù)據(jù)   complete() { console.log('Finished sequence'); } });

運行結果如下：

上面一個例子演示了如何自定義訂閱函數(shù)，那么既然可以自定義訂閱者函數(shù)，我們就可以將異步代碼封裝進可觀察對象的訂閱者函數(shù)中，待異步代碼執(zhí)行完再發(fā)送數(shù)據(jù)。如下：

import { Observable } from 'rxjs' // 異步函數(shù) function fn(num) {     return new Promise((reslove, reject) => {         setTimeout(() => {             num++             reslove(num)         }, 1000)     }) } // 創(chuàng)建可觀察對象，并傳入訂閱者函數(shù) const observable = new Observable((x) => {     let num = 1     fn(num).then(     	res => x.next(res)	// 異步代碼執(zhí)行完成，發(fā)送數(shù)據(jù)     )  }) // 訂閱，接收數(shù)據(jù)，可以改為鏈式調用 observable.subscribe(data => console.log(data))	// 2

多播

https://angular.cn/guide/observables#multicasting

RxJS操作符

我們可以使用一系列的RxJS操作符，在這些消息被接收方接收之前，對它們進行一系列的處理、轉換，因為這些操作符都是純函數(shù)。

import { of } from 'rxjs'; import { map } from 'rxjs/operators'; // 1、創(chuàng)建可觀察對象，并發(fā)送數(shù)據(jù) const nums = of(1, 2, 3); // 2、創(chuàng)建函數(shù)以接受可觀察對象 const squareValues = map((val: number) => val * val); const squaredNums = squareValues(nums);  squaredNums.subscribe(x => console.log(x));

上面的方式我看不懂且難以接受，一般常用下面這種，使用pipe把多個操作符鏈接起來

import { map, Observable, filter } from 'rxjs'  // 創(chuàng)建可觀察對象，并傳入訂閱者函數(shù) const observable = new Observable((x) => {     x.next(1)     x.next(2)     x.next(3)     x.next(4) }).pipe(     map(value => value*100),		// 操作符     filter(value => value == 200)	// 操作符 ) .subscribe(data => console.log(data))	// 200

錯誤處理

RxJS 還提供了catchError 操作符，它允許你在管道中處理已知錯誤。
假設你有一個可觀察對象，它發(fā)起 API 請求，然后對服務器返回的響應進行映射。如果服務器返回了錯誤或值不存在，就會生成一個錯誤。如果你捕獲這個錯誤并提供了一個默認值，流就會繼續(xù)處理這些值，而不會報錯。如下：

import { map, Observable, filter, catchError, of } from 'rxjs'  const observable = new Observable((x) => {         x.next(1)	// 發(fā)送數(shù)據(jù) 1 和 2         x.next(2) }).pipe(     map(value => {         if (value === 1) {	// 1、當發(fā)送的數(shù)據(jù)為 1 時，將其乘以 100             return value*100         } else {	// 2、否則拋出錯誤             throw new Error('拋出錯誤');         }     }),     // 3、此處捕獲錯誤并處理錯誤，對外發(fā)送數(shù)據(jù) 0     catchError((err) => {         console.log(err)         return of(0)     }) ) .subscribe(     data => console.log(data),     // 4、由于上面拋出的錯誤被 catchError 操作符處理（重新發(fā)送數(shù)據(jù)）了，所以這里能順利訂閱到數(shù)據(jù)而不報錯     err => console.log('接受不到數(shù)據(jù)：', err) )

最后的運行結果如下：