第淺析Angular中的可觀察對(duì)象、觀察者和RxJS操作符只有當(dāng)有人訂閱Observable的實(shí)例時(shí)，它才會(huì)開始發(fā)布值。訂閱時(shí)要先調(diào)用可觀察對(duì)象的subscribe()方法，并把一個(gè)觀察者對(duì)象傳給它，用來接收通知。如下：

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

of(...items)：返回一個(gè)Observable實(shí)例，它用同步的方式把參數(shù)中提供的這些值一個(gè)一個(gè)發(fā)送出來。

from(iterable)：把它的參數(shù)轉(zhuǎn)換成一個(gè)Observable實(shí)例。該方法通常用于把一個(gè)數(shù)組轉(zhuǎn)換成一個(gè)（發(fā)送多個(gè)值的）可觀察對(duì)象。

import{of}fromrxjs

//1、通過of()方法返回一個(gè)可觀察對(duì)象，并準(zhǔn)備將1，2，3三個(gè)數(shù)據(jù)發(fā)送出去

constobservable=of(1,2,3);

//2、實(shí)現(xiàn)observer接口，觀察者

constobserver={

next:(num:number)=console.log(num),

error:(err:Error)=console.error(Observergotanerror:+err),

complete:()=console.log(Observergotacompletenotification),

//3、訂閱。調(diào)用可觀察對(duì)象的subscribe()方法訂閱，subscribe()方法中傳入的對(duì)象就是一個(gè)觀察者

observable.subscribe(observer);

運(yùn)行結(jié)果如下：

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

observable.subscribe(

num=console.log(num),

err=console.error(Observergotanerror:+err),

()=console.log(Observergotacompletenotification)

);

訂閱者函數(shù)

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

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

//1、自定義訂閱者函數(shù)

functionsequenceSubscriber(observer:Observernumber){

observer.next(1);//發(fā)送數(shù)據(jù)

observer.next(2);//發(fā)送數(shù)據(jù)

observer.next(3);//發(fā)送數(shù)據(jù)

plete();

return{unsubscribe(){}};

//2、通過構(gòu)造函數(shù)創(chuàng)建一個(gè)新的可觀察對(duì)象，參數(shù)就是一個(gè)訂閱者函數(shù)

constsequence=newObservable(sequenceSubscriber);

//3、訂閱

sequence.subscribe({

next(num){console.log(num);},//接受數(shù)據(jù)

complete(){console.log(Finishedsequence}

});

運(yùn)行結(jié)果如下：

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

import{Observable}fromrxjs

//異步函數(shù)

functionfn(num){

returnnewPromise((reslove,reject)={

setTimeout(()={

num++

reslove(num)

},1000)

//創(chuàng)建可觀察對(duì)象，并傳入訂閱者函數(shù)

constobservable=newObservable((x)={

letnum=1

fn(num).then(

res=x.next(res)//異步代碼執(zhí)行完成，發(fā)送數(shù)據(jù)

//訂閱，接收數(shù)據(jù)，可以改為鏈?zhǔn)秸{(diào)用

observable.subscribe(data=console.log(data))//2

/guide/observables#multicasting

RxJS操作符

我們可以使用一系列的RxJS操作符，在這些消息被接收方接收之前，對(duì)它們進(jìn)行一系列的處理、轉(zhuǎn)換，因?yàn)檫@些操作符都是純函數(shù)。

import{of}fromrxjs

import{map}fromrxjs/operators

//1、創(chuàng)建可觀察對(duì)象，并發(fā)送數(shù)據(jù)

constnums=of(1,2,3);

//2、創(chuàng)建函數(shù)以接受可觀察對(duì)象

constsquareValues=map((val:number)=val*val);

constsquaredNums=squareValues(nums);

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

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

import{map,Observable,filter}fromrxjs

//創(chuàng)建可觀察對(duì)象，并傳入訂閱者函數(shù)

constobservable=newObservable((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操作符，它允許你在管道中處理已知錯(cuò)誤。

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

import{map,Observable,filter,catchError,of}fromrxjs

constobservable=newObservable((x)={

x.next(1)//發(fā)送數(shù)據(jù)1和2

x.next(2)

}).pipe(

map(value={

if(value===1){//1、當(dāng)發(fā)送的數(shù)據(jù)為1時(shí)，將其乘以100

returnvalue*100

}else{//2、否則拋出錯(cuò)誤

thrownewError(拋出錯(cuò)誤

//3、此處捕獲錯(cuò)誤并處理錯(cuò)誤，對(duì)外發(fā)送數(shù)據(jù)0

catchError((err)={

console.log(err)