RxJS 核心概念之Subject

weij 發(fā)布于2019-08-19 18:15 / 2521人閱讀

摘要：返回的對(duì)象同時(shí)是類型的，擁有方法。由于調(diào)用后，開始執(zhí)行，因此，會(huì)返回一個(gè)供調(diào)用者來終止執(zhí)行。是的一個(gè)衍生類，具有最新的值的概念。舉一個(gè)形象的例子，表示一個(gè)人的生日，而則表示一個(gè)人的歲數(shù)。

什么是Subject？ 在RxJS中，Subject是一類特殊的Observable，它可以向多個(gè)Observer多路推送數(shù)值。普通的Observable并不具備多路推送的能力（每一個(gè)Observer都有自己獨(dú)立的執(zhí)行環(huán)境），而Subject可以共享一個(gè)執(zhí)行環(huán)境。

Subject是一種可以多路推送的可觀察對(duì)象。與EventEmitter類似，Subject維護(hù)著自己的Observer。

每一個(gè)Subject都是一個(gè)Observable（可觀察對(duì)象） 對(duì)于一個(gè)Subject，你可以訂閱（subscribe）它，Observer會(huì)和往常一樣接收到數(shù)據(jù)。從Observer的視角看，它并不能區(qū)分自己的執(zhí)行環(huán)境是普通Observable的單路推送還是基于Subject的多路推送。

Subject的內(nèi)部實(shí)現(xiàn)中，并不會(huì)在被訂閱（subscribe）后創(chuàng)建新的執(zhí)行環(huán)境。它僅僅會(huì)把新的Observer注冊(cè)在由它本身維護(hù)的Observer列表中，這和其他語言、庫(kù)中的addListener機(jī)制類似。

每一個(gè)Subject也可以作為Observer（觀察者） Subject同樣也是一個(gè)由next(v)，error(e)，和 complete()這些方法組成的對(duì)象。調(diào)用next(theValue)方法后，Subject會(huì)向所有已經(jīng)在其上注冊(cè)的Observer多路推送theValue。

下面的例子中，我們?cè)赟ubject上注冊(cè)了兩個(gè)Observer，并且多路推送了一些數(shù)值：

var subject = new Rx.Subject();

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});
subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(1);
subject.next(2);

控制臺(tái)輸出結(jié)果如下：

observerA: 1
observerB: 1
observerA: 2
observerB: 2

既然Subject是一個(gè)Observer，你可以把它作為subscribe（訂閱）普通Observable時(shí)的參數(shù)，如下面例子所示：

var subject = new Rx.Subject();

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});
subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

var observable = Rx.Observable.from([1, 2, 3]);

observable.subscribe(subject); // 你可以傳遞Subject來訂閱observable

執(zhí)行后結(jié)果如下：

observerA: 1
observerB: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3

通過上面的實(shí)現(xiàn)：我們發(fā)現(xiàn)可以通過Subject將普通的Observable單路推送轉(zhuǎn)換為多路推送。這說明了Subject的作用——作為單路Observable轉(zhuǎn)變?yōu)槎嗦稯bservable的橋梁。

還有幾種特殊的Subject 類型，分別是BehaviorSubject，ReplaySubject，和 AsyncSubject。

多路推送的Observable

在以后的語境中，每當(dāng)提到“多路推送的Observable”，我們特指通過Subject構(gòu)建的Observable執(zhí)行環(huán)境。否則“普通的Observable”只是一個(gè)不會(huì)共享執(zhí)行環(huán)境并且被訂閱后才生效的一系列值。

通過使用Subject可以創(chuàng)建擁有相同執(zhí)行環(huán)境的多路的Observable。

下面展示了多路的運(yùn)作方式：Subject從普通的Observable訂閱了數(shù)據(jù)，然后其他Observer又訂閱了這個(gè)Subject，示例如下：

var source = Rx.Observable.from([1, 2, 3]);
var subject = new Rx.Subject();
var multicasted = source.multicast(subject);

// 通過`subject.subscribe({...})`訂閱Subject的Observer：
multicasted.subscribe({
  next: (v) => console.log("observerA: " + v)
});
multicasted.subscribe({
  next: (v) => console.log("observerB: " + v)
});

// 讓Subject從數(shù)據(jù)源訂閱開始生效：
multicasted.connect();

multicast方法返回一個(gè)類似于Observable的可觀察對(duì)象，但是在其被訂閱后，它會(huì)表現(xiàn)Subject的特性。 multicast 返回的對(duì)象同時(shí)是ConnectableObservable類型的，擁有connect() 方法。

connect()方法非常的重要，它決定Observable何時(shí)開始執(zhí)行。由于調(diào)用connect()后，Observable開始執(zhí)行，因此，connect()會(huì)返回一個(gè)Subscription供調(diào)用者來終止執(zhí)行。

引用計(jì)數(shù)

通過手動(dòng)調(diào)用connect()返回的Subscription控制執(zhí)行十分繁雜。通常，我們希望在有第一個(gè)Observer訂閱Subject后自動(dòng)connnect，當(dāng)所有Observer都取消訂閱后終止這個(gè)Subject。

我們來分析一下下面例子中subscription的過程：

第一個(gè)Observer 訂閱了多路推送的 Observable

多路Observable被連接

向第一個(gè)Observer發(fā)送值為0的next通知

第二個(gè)Observer訂閱了多路推送的 Observable

向第一個(gè)Observer發(fā)送值為1的next通知

向第二個(gè)Observer發(fā)送值為1的next通知

第一個(gè)Observer取消了對(duì)多路推送的Observable的訂閱

向第二個(gè)Observer發(fā)送值為2的next通知

第二個(gè)Observer取消了對(duì)多路推送的Observable的訂閱

取消對(duì)多路推送的Observable的連接

通過顯式地調(diào)用connect()，代碼如下：

var source = Rx.Observable.interval(500);
var subject = new Rx.Subject();
var multicasted = source.multicast(subject);
var subscription1, subscription2, subscriptionConnect;

subscription1 = multicasted.subscribe({
  next: (v) => console.log("observerA: " + v)
});
subscriptionConnect = multicasted.connect();

setTimeout(() => {
  subscription2 = multicasted.subscribe({
    next: (v) => console.log("observerB: " + v)
  });
}, 600);

setTimeout(() => {
  subscription1.unsubscribe();
}, 1200);

setTimeout(() => {
  subscription2.unsubscribe();
  subscriptionConnect.unsubscribe(); 
}, 2000);

如果你不想顯式地調(diào)用connect()方法，可以在ConnectableObservable類型的Observable上調(diào)用refCount()方法。方法會(huì)進(jìn)行引用計(jì)數(shù)：記錄Observable被訂閱的行為。當(dāng)訂閱數(shù)從 0 到 1時(shí)refCount() 會(huì)調(diào)用connect() 方法。到訂閱數(shù)從1 到 0，他會(huì)終止整個(gè)執(zhí)行過程。

refCount 使得多路推送的Observable在被訂閱后自動(dòng)執(zhí)行，在所有觀察者取消訂閱后，停止執(zhí)行。

下面是示例：

var source = Rx.Observable.interval(500);
var subject = new Rx.Subject();
var refCounted = source.multicast(subject).refCount();
var subscription1, subscription2, subscriptionConnect;

console.log("observerA subscribed");
subscription1 = refCounted.subscribe({
  next: (v) => console.log("observerA: " + v)
});

setTimeout(() => {
  console.log("observerB subscribed");
  subscription2 = refCounted.subscribe({
    next: (v) => console.log("observerB: " + v)
  });
}, 600);

setTimeout(() => {
  console.log("observerA unsubscribed");
  subscription1.unsubscribe();
}, 1200);

setTimeout(() => {
  console.log("observerB unsubscribed");
  subscription2.unsubscribe();
}, 2000);

執(zhí)行輸出結(jié)果如下:

observerA subscribed
observerA: 0
observerB subscribed
observerA: 1
observerB: 1
observerA unsubscribed
observerB: 2
observerB unsubscribed

只有ConnectableObservables擁有refCount()方法，調(diào)用后會(huì)返回一個(gè)Observable而不是新的ConnectableObservable。

BehaviorSubject

BehaviorSubject是Subject的一個(gè)衍生類，具有“最新的值”的概念。它總是保存最近向數(shù)據(jù)消費(fèi)者發(fā)送的值，當(dāng)一個(gè)Observer訂閱后，它會(huì)即刻從BehaviorSubject收到“最新的值”。

BehaviorSubjects非常適于表示“隨時(shí)間推移的值”。舉一個(gè)形象的例子，Subject表示一個(gè)人的生日，而Behavior則表示一個(gè)人的歲數(shù)。（生日只是一天，一個(gè)人的歲數(shù)會(huì)保持到下一次生日之前。）

下面例子中，展示了如何用 0初始化BehaviorSubject，當(dāng)Observer訂閱它時(shí)，0是第一個(gè)被推送的值。緊接著，在第二個(gè)Observer訂閱BehaviorSubject之前，它推送了2，雖然訂閱在推送2之后，但是第二個(gè)Observer仍然能接受到2：

var subject = new Rx.BehaviorSubject(0 /* 初始值 */);

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);

subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(3);

輸出結(jié)果如下：

observerA: 0
observerA: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3

ReplaySubject

ReplaySubject 如同于BehaviorSubject是 Subject 的子類。通過 ReplaySubject可以向新的訂閱者推送舊數(shù)值，就像一個(gè)錄像機(jī)ReplaySubject可以記錄Observable的一部分狀態(tài)（過去時(shí)間內(nèi)推送的值）。

.一個(gè)ReplaySubject可以記錄Observable執(zhí)行過程中推送的多個(gè)值，并向新的訂閱者回放它們。

你可以指定回放值的數(shù)量：

var subject = new Rx.ReplaySubject(3 /* 回放數(shù)量 */);

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(5);

輸出如下：

observerA: 1
observerA: 2
observerA: 3
observerA: 4
observerB: 2
observerB: 3
observerB: 4
observerA: 5
observerB: 5

除了回放數(shù)量，你也可以以毫秒為單位去指定“窗口時(shí)間”，決定ReplaySubject記錄多久以前Observable推送的數(shù)值。下面的例子中，我們把回放數(shù)量設(shè)置為100，把窗口時(shí)間設(shè)置為500毫秒：

var subject = new Rx.ReplaySubject(100, 500 /* windowTime */);

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});

var i = 1;
setInterval(() => subject.next(i++), 200);

setTimeout(() => {
  subject.subscribe({
    next: (v) => console.log("observerB: " + v)
  });
}, 1000);

第二個(gè)Observer接受到3（600ms）, 4（800ms）和 5（1000ms），這些值均在訂閱之前的500毫秒內(nèi)推送（窗口長(zhǎng)度 1000ms - 600ms = 400ms < 500ms）：

observerA: 1
observerA: 2
observerA: 3
observerA: 4
observerA: 5
observerB: 3
observerB: 4
observerB: 5
observerA: 6
observerB: 6
...

AsyncSubject

AsyncSubject是Subject的另外一個(gè)衍生類，Observable僅會(huì)在執(zhí)行完成后，推送執(zhí)行環(huán)境中的最后一個(gè)值。

var subject = new Rx.AsyncSubject();

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(5);
subject.complete();

輸出結(jié)果如下：