可觀察的（Observable）

可觀察集合（Observables）是多值懶推送集合。它們填補(bǔ)了下面表格的空白：

	SINGLE	MULTIPLE
Pull	Function	Iterator
Push	Promise	Observable

舉個(gè)例子，下面是一個(gè)可觀察對(duì)象，當(dāng)它被訂閱的時(shí)候立即推送值 1,2,3，并且值 4 在訂閱調(diào)用之后的一秒傳遞過(guò)去，然后完成：

import { Observable } from 'rxjs';

const Observable = new Observable(subscriber => {
    subscripber.next(1);
    subscripber.next(2);
    subscripber.next(3);
    setTimeout(() => {
        subscripber.next(4);
        subscripber.complete();
    }, 1000);
});

為了調(diào)用 Observable 以及看到這些值，我們需要訂閱它：

import { Observable } from 'rxjs';

const observable = new Observable(subscriber => {
    subscripber.next(1);
    subscripber.next(2);
    subscripber.next(3);
    setTimeout(() => {
        subscripber.next(4);
        subscripber.complete();
    }, 1000);
});

console.log('just before subscribe');
observable.subscribe({
    next(x) { console.log('got value '+ x); },
    error(err) { console.log('something wrong occurred: ' + err); },
    complete() { console.log('done'); }
});
console.log('just after subscribe');

它執(zhí)行會(huì)打印如下內(nèi)容：

just before subscribe
got value 1
got value 2
got value 3
just after subscribe
got value 4
done

Pull vs Push

Pull 和 Push 是兩個(gè)不同的協(xié)議，他們描述了數(shù)據(jù)生產(chǎn)者如何與數(shù)據(jù)消費(fèi)者通信。

什么是 Pull？在 Pull 系統(tǒng)，消費(fèi)者決定何時(shí)從數(shù)據(jù)生產(chǎn)者接收數(shù)據(jù)。生產(chǎn)者自己也不知道數(shù)據(jù)什么時(shí)候推送給消費(fèi)者。所有的 javascript 函數(shù)都是一個(gè) Pull 系統(tǒng)。函數(shù)是數(shù)據(jù)的生產(chǎn)者，并且這個(gè)調(diào)用函數(shù)的代碼通過(guò)從它調(diào)用中“提取（pulling）”一個(gè)返回值來(lái)消耗數(shù)據(jù)。

ES2015 介紹了生成器函數(shù)和迭代器 （function* ）其他類(lèi)型的 Pull 系統(tǒng)。調(diào)用 iterator.next 的代碼是消費(fèi)者，從迭代器（生產(chǎn)者）中 “提取” 出多個(gè)值。

	Producer	Consumer
Pull	被動(dòng)：請(qǐng)求時(shí)生成數(shù)據(jù)	主動(dòng)：決定數(shù)據(jù)何時(shí)請(qǐng)求
Push	主動(dòng)：在自己的空間生成數(shù)據(jù)	被動(dòng)：響應(yīng)接收到的數(shù)據(jù)

什么是 Push？Push 系統(tǒng)，生成者決定何時(shí)發(fā)送數(shù)據(jù)給消費(fèi)者。消費(fèi)者是不知道數(shù)據(jù)是何時(shí)接收的。

在如今的 javascript 中 Pull 系統(tǒng)中最通用的類(lèi)型是 Promises。一個(gè) Promise（生產(chǎn)者）派送一個(gè)被解析的值到已經(jīng)注冊(cè)的回調(diào)函數(shù)（消費(fèi)者）中，但是不像函數(shù)，Promise 它負(fù)責(zé)準(zhǔn)確的確定這個(gè)值何時(shí)被 “推送” 到回調(diào)函數(shù)。

RxJS 介紹了 Observables，新的 Push 系統(tǒng)。一個(gè)可觀察的對(duì)象是多個(gè)值的生產(chǎn)者，“推送” 他們到 Observables（消費(fèi)者）。

函數(shù)是一種延遲計(jì)算，調(diào)用同步返回一個(gè)值
生成器是一種延遲計(jì)算，在迭代中同步返回 0 到 無(wú)窮值（可能的）。
Promise 是一個(gè)計(jì)算，它可能（也有能不會(huì)）最后返回一個(gè)值
Observable 是一個(gè)延遲計(jì)算的，它能同步或異步返回 0 到 無(wú)窮值（可能）從它們被調(diào)用的那刻開(kāi)始

Observables 函數(shù)一般化

與流行的說(shuō)法相反，Observables 不像事件驅(qū)動(dòng)器那樣，也不像多個(gè)值的 Promise。Observables 也許行為像事件驅(qū)動(dòng)器在有些例子中，即在他們使用 RxJS Subjects 多播時(shí)，通常它們不像事件驅(qū)動(dòng)。

Observables 像一個(gè)無(wú)參的函數(shù)，但是它們?cè)试S生成多個(gè)值。

考慮下面代碼：

function foo()
{
    console.log('Hello');
    return 42;
}

const x = foo.call();	//same as foo()
console.log(x);
const y = foo.call();
console.log(y);

我們期望能看到如下輸出：

"Hello"
42
"Hello"
42

你可以根據(jù)上面的行為用 Observables 寫(xiě)相同的行為：

import { Observable } from 'rxjs';
const foo = new Observable(subscriber => {
    console.log('Hello');
    subscriber.next(42);
});

foo.subscriber(x => {
    console.log(x);
});
foo.subscriber(y => {
    console.log(y);
});

同樣會(huì)有相同的輸出：

"Hello"
42
"Hello"
42

這兩個(gè)函數(shù)發(fā)生的都是延遲計(jì)算。如果你不調(diào)用函數(shù)，console.log('Hello') 是不會(huì)觸發(fā)的。Observables 同樣也是如此，如果你不 “call” 它（通過(guò) subscribe）console.log('Hello') 將不會(huì)觸發(fā)。另外，“calling” 或者 “subscribing” 是一個(gè)隔離的操作：兩個(gè)函數(shù)調(diào)用觸發(fā)兩個(gè)獨(dú)立的副作用，并且兩個(gè) Observable 訂閱觸發(fā)兩個(gè)副作用。與此相反，事件驅(qū)動(dòng)器它會(huì)共享副作用，并且無(wú)論訂閱存不存在都會(huì)執(zhí)行，Observable 不會(huì)共享執(zhí)行并都是延遲的。

訂閱（Subscribing）一個(gè)可觀察的對(duì)象（Observable）跟調(diào)用一個(gè)函數(shù)是類(lèi)似的。

Observable 和 函數(shù)之間的區(qū)別是什么呢？Observable 能隨著時(shí)間返回多個(gè)值，還有一些是函數(shù)不能做的。比如你不能這么做：

function foo()
{
    console.log('hello');
    return 42;
    return 100;	//死代碼。永遠(yuǎn)不會(huì)發(fā)生這行
}

函數(shù)只能返回一個(gè)值。Observable 就能做到這個(gè)：

import { Observable } from 'rxjs';

const foo = new Observable(subscribe => {
    console.log('Hello');
    subscriber.next(42);
    subscriber.next(100);	//返回值
    subscriber.next(200);	//同樣還是返回值
})

console.log('before');
foo.subscribe(x => {
    console.log(x);
});
console.log('after');

同步輸出：

"before"
"before"
42
100
200
"after"

你也能異步返回這些值：

import { Observable } from 'rxjs';

const foo = new Observable(subscriber => {
    console.log('hello');
    subscriber.next(42);
  	subscriber.next(100);
  	subscriber.next(200);
    setTimeout(() => {
        subscriber.next(300);	//異步
    },1000);
});

console.log('before');
foo.subscribe(x => {
  console.log(x);
});
console.log('after');

輸出：

"before"
"Hello"
42
100
200
"after"
300

結(jié)論：

func.call() 意思是說(shuō) “同步給我一個(gè)值”
observable.subscribe() 意思是說(shuō) “給我任何數(shù)值，不管同步還是異步”

深入解析 Observable

Observables 是通過(guò) new Observable 或一個(gè)創(chuàng)建操作符建立的，由觀察者訂閱，執(zhí)行去推送 next / error / complete 通知到觀察者（Observer），并且他們的執(zhí)行也許會(huì)被處理。這里有 4 個(gè)方面都編碼在可觀察的（Observable）實(shí)例對(duì)象中，但是有一些方面與其他類(lèi)型相關(guān)，比如 Observer 和 Subscription。

核心的 Observable 概念如下：

創(chuàng)建可觀察的（Observables）對(duì)象
訂閱 Observables
執(zhí)行 Observables
釋放 Observables

創(chuàng)建 Observables

Observable 有個(gè)帶一個(gè)參數(shù)的構(gòu)造函數(shù)：subscribe 函數(shù)。

下面的例子是創(chuàng)建一個(gè) Observable，來(lái)向訂閱者每秒發(fā)送一個(gè)字符串 ‘hi’。

import { Observable } from 'rxjs';

const observable = new Observable(subscribe => {
    const id = setInterval(() => {
        subscribe.next('hi')
    },1000);
});

Observables 能通過(guò)實(shí)例化創(chuàng)建。但是多數(shù)情況下，observables 都是通過(guò)使用創(chuàng)建函數(shù)創(chuàng)建的，比如 of,from,interval 等。

在上面的例子，subscribe 函數(shù)部分來(lái)訂閱 Observable 是最重要的。讓我們來(lái)看看訂閱者訂閱的意思。

訂閱 Observables

Observable 的實(shí)例對(duì)象 observable 在例子中能夠被訂閱，像這樣：

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

observable.subscribe 和 在實(shí)例化 new Observable(function subscribe(subscriber){...}) 中的 subscribe 有相同的名字這不是巧合。在庫(kù)中，它們是不同的，出于特定目的你可以考慮他們的概念是一致的。

這個(gè)展示了 subscribe 怎樣調(diào)用是不會(huì)在多個(gè)可觀察對(duì)象的相同可觀察對(duì)象之間共享。當(dāng)觀察者 Observer 調(diào)用 observable.subscribe，new Observable(function subscribe(subscriber) {...}) 中的函數(shù) subscribe 運(yùn)行在給定的訂閱器上運(yùn)行。

訂閱一個(gè)可觀察者對(duì)象（Observable）就像調(diào)用一個(gè)函數(shù)，在數(shù)據(jù)將被傳遞到的地方提供一個(gè)回調(diào)函數(shù)。

這對(duì)于事件處理 API 來(lái)說(shuō)是徹底不同的，像 addEventListener / removeEventListener。通過(guò) observable.subscribe,給定的觀察者是不會(huì)作為偵聽(tīng)者注冊(cè)到可觀察對(duì)象（Observable）中去。Observable 甚至不會(huì)維護(hù)已經(jīng)附加進(jìn)來(lái)的列表。

調(diào)用 subscribe 是一種簡(jiǎn)單的方式來(lái)開(kāi)始一個(gè) “可觀察的執(zhí)行” 以及傳遞值或是傳遞執(zhí)行的可觀察的事件。

執(zhí)行 Observables

在代碼 new Observable(function subscribe(subscriber){...}) 代碼里表示一個(gè) "可觀察的執(zhí)行"，是一個(gè)延遲執(zhí)行，它只發(fā)生在每個(gè)觀察者（Observer）訂閱的時(shí)候。這個(gè)執(zhí)行多次會(huì)生成多個(gè)值，無(wú)論同步還是異步。

這里有三個(gè)類(lèi)型的值在可觀察者執(zhí)行時(shí)傳遞：

"Next" 通知：發(fā)送一個(gè)值，比如數(shù)字類(lèi)型，字符串類(lèi)型以及對(duì)象等
"Error" 通知：發(fā)送一個(gè) Javascript 錯(cuò)誤或異常
"Complete" 通知：不會(huì)發(fā)送任何值

"Next" 通知是最重要的并且是最常用的類(lèi)型：它們表示一個(gè)數(shù)據(jù)被傳遞到訂閱者的行為。"Error" 和 "Complete" 通知可能只發(fā)生一次在可觀察的對(duì)象執(zhí)行期間，它們只能發(fā)生一個(gè)。

在調(diào)用 Observable 語(yǔ)法或契約中，這些約束表現(xiàn)得最好，寫(xiě)成一個(gè)正則表達(dá)式：

next*(error|complete)?

在 Observable 執(zhí)行中，下一個(gè)通知可以從 0 發(fā)送到無(wú)窮。如一個(gè) Error 或是 Complete 被發(fā)送了，那么在之后就不會(huì)有任何東西被發(fā)送：

import { Observable } from 'rxjs';

const observable = new Observable(function subscribe(subscriber){
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    subscriber.complete();
});

Observables 嚴(yán)格遵守 Observable 契約，所以下面這段代碼將不會(huì)傳遞到 Next 通知 4：

import { Observable } from 'rxjs';

const observable = new Observable(function subscribe(subscriber){
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    subscriber.complete();
    subscriber.next(4);//這里永遠(yuǎn)都不會(huì)執(zhí)行。因?yàn)橐呀?jīng)complete
})

這種方法是好的，在 subscriber 中通過(guò) try/catch 代碼塊包裝任何代碼，它將傳遞一個(gè) Error 通知，如果捕捉到異常的話。

import { Observable } from 'rxjs';

const observable = new Observable(function subscribe(subscriber){
    try{
        subscriber.next(1);
        subscriber.next(2);
        subscriber.next(3);
        subscriber.complete();
    } catch (err) {
        subscriber.error(err);	//傳遞一個(gè)錯(cuò)誤對(duì)象，如果捕捉到異常的話。
    }
});

釋放 Observable 執(zhí)行

由于 Observaber 執(zhí)行可能會(huì)無(wú)窮，并且共同的就是觀察者（Observer）想要在有限的時(shí)間放棄這個(gè)執(zhí)行，所以我們需要一個(gè) api 來(lái)取消執(zhí)行。由于每個(gè)執(zhí)行在一個(gè) Observer 都是獨(dú)有的，一旦 Observer 做完了接收數(shù)據(jù)，為了避免浪費(fèi)計(jì)算以及內(nèi)存資源，那么它就必須有一種方式來(lái)停止執(zhí)行。

當(dāng) observable.subscribe 被調(diào)用時(shí)，觀察者（Observer）被附加到最新創(chuàng)建的 Observable 執(zhí)行。Subscription 它也返回一個(gè)對(duì)象：

const subscription = observable.subscribe(x => console.log(x));

Subscription 表示正在執(zhí)行，有一個(gè)最小的 api，這個(gè) api 允許你取消執(zhí)行。關(guān)于 Subscription 這里可以閱讀更多。通過(guò) subscription.unsubcribe() 你可以取消正在執(zhí)行中的執(zhí)行。

import { from } from 'rxjs';

const observable = from([10,20,30]);
const subscription = observable.subscribe(x => console.log(x));
//之后
subscribe.unsubscribe();

當(dāng)你訂閱的時(shí)候，你會(huì)得到一個(gè) Subscription，它表示正在執(zhí)行的執(zhí)行。僅僅調(diào)用 unsubscribe 取消執(zhí)行。

每個(gè) Observable 必須定義怎么去釋放執(zhí)行的資源，當(dāng)我們使用 create 創(chuàng)建 Observable 時(shí)。你可以通過(guò)返回一個(gè)自定義的 ubsubcribe 方法做到這個(gè)。

舉個(gè)例子，這里展示了我們通過(guò) setInterval 清除一個(gè)間隔執(zhí)行：

const observable = new Observable(function subscribe(subscriber){
    //根據(jù)間隔資源
    const intervalId = setInterval(() => {
        subscribe,next('hi')
    },1000);
    
    //提供一個(gè)方法來(lái)取消以及釋放間隔資源。
    return function unsubscribe()
    {
        clearInterval(intervaleId);
    }
})

由此可見(jiàn) observable.subscribe 相似 new Observable(function subscribe(subscriber){...})，我們從 subscribe 返回 unsubscribe，在概念上等同于 subscribe.unsubscribe。實(shí)際上，如果我們移除圍繞 ReactiveX 概念的類(lèi)型，剩下的 JavaScript 代碼就不在陌生了。

function subscribe(subscriber)
{
    const intervalId = setInterval(() => {
        subscribe.next('hi');
    },1000);
    
    return function unsubscribe(){
        clearInterval(intervalId);
    }
}

const unsubscribe = subscribe({next: (x) => console.log(x)});

//最后
unsubscriber();	//釋放資源

我們使用的 RxJS 類(lèi)型，如 Observable，Observer，Subscription 的理由是更安全的獲得（例如 Obscervable 契約）以及 Operators（操作符）的組合性。

總結(jié)

以上是生活随笔為你收集整理的RxJS——可观察的对象（Observable）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。