摘要：響應(yīng)式編程第一章響應(yīng)式響應(yīng)式編程第二章序列的深入研究響應(yīng)式編程第三章構(gòu)建并發(fā)程序響應(yīng)式編程第四章構(gòu)建完整的應(yīng)用程序響應(yīng)式編程第五章使用管理時(shí)間響應(yīng)式編程第六章使用的響應(yīng)式應(yīng)用程序使用管理時(shí)間自從接觸，就開(kāi)始在我的項(xiàng)目中使用它。

Rxjs 響應(yīng)式編程-第一章：響應(yīng)式
Rxjs 響應(yīng)式編程-第二章：序列的深入研究
Rxjs 響應(yīng)式編程-第三章: 構(gòu)建并發(fā)程序
Rxjs 響應(yīng)式編程-第四章 構(gòu)建完整的Web應(yīng)用程序
Rxjs 響應(yīng)式編程-第五章 使用Schedulers管理時(shí)間
Rxjs 響應(yīng)式編程-第六章 使用Cycle.js的響應(yīng)式Web應(yīng)用程序

自從接觸RxJS，就開(kāi)始在我的項(xiàng)目中使用它。有一段時(shí)間我以為我知道如何有效地使用它，但有一個(gè)令人煩惱的問(wèn)題：我怎么知道我使用的運(yùn)算符是同步還是異步？換句話說(shuō)，Operators到底什么時(shí)候發(fā)出通知？這似乎是正確使用RxJS的關(guān)鍵部分，但對(duì)我來(lái)說(shuō)感覺(jué)有點(diǎn)模糊。

我認(rèn)為，間隔運(yùn)算符顯然是異步的，所以它在內(nèi)部使用類(lèi)似setTimeout的東西來(lái)發(fā)出項(xiàng)目。但是，如果我使用范圍怎么辦？它也是異步發(fā)射的嗎？它會(huì)阻止事件循環(huán)嗎？來(lái)自哪里？我到處都在使用這些運(yùn)算符，但我對(duì)它們的內(nèi)部并發(fā)模型知之甚少。

然后我了解了Schedulers。

Schedulers是一種強(qiáng)大的機(jī)制，可以精確管理應(yīng)用程序中的并發(fā)性。它們?cè)试S您隨時(shí)更改其并發(fā)模型，從而對(duì)Observable如何發(fā)出通知進(jìn)行細(xì)粒度控制。在本章中，您將學(xué)習(xí)如何使用調(diào)度程序并在常見(jiàn)場(chǎng)景中應(yīng)用它們。我們將專(zhuān)注于測(cè)試，調(diào)度程序特別有用，您將學(xué)習(xí)如何制作自己的Schedulers。

Schedulers是一種“安排”將來(lái)發(fā)生的操作的機(jī)制。 RxJS中的每個(gè)運(yùn)算符在內(nèi)部使用一個(gè)Schedulers，選擇該Schedulers以在最可能的情況下提供最佳性能。

讓我們看看我們?nèi)绾胃淖冞\(yùn)算符中的Schedulers以及這樣做的后果。 首先讓我們創(chuàng)建一個(gè)包含1,000個(gè)整數(shù)的數(shù)組：

var arr = [];
for (var i=0; i<1000; i++) {
    arr.push(i);
}

然后，我們從arr創(chuàng)建一個(gè)Observable并強(qiáng)制它通過(guò)訂閱它來(lái)發(fā)出所有通知。 在代碼中，我們還保存了發(fā)出所有通知所需的時(shí)間：

var timeStart = Date.now();
Rx.Observable.from(arr).subscribe(
    function onNext() {},
    function onError() {},
    function onCompleted() {
        console.log("Total time: " + (Date.now() - timeStart) + "ms");
    }
);

"Total time: 6ms”

六毫秒 - 不壞！ from在內(nèi)部使用Rx.Scheduler.currentThread，它計(jì)劃在任何當(dāng)前工作完成后運(yùn)行。 一旦啟動(dòng)，它將同步處理所有通知。

在讓我們將Scheduler更改為Rx.Scheduler.default

var timeStart = Date.now();
Rx.Observable.from(arr, null, null, Rx.Scheduler.default).subscribe(
    function onNext() {},
    function onError() {},
    function onCompleted() {
        console.log("Total time: " + (Date.now() - timeStart) + "ms");
    }
);

"Total time: 5337ms”

哇，我們的代碼運(yùn)行速度比使用currentThread Scheduler慢幾千倍。 那是因?yàn)槟J(rèn)的Scheduler異步運(yùn)行每個(gè)通知。 我們可以通過(guò)在訂閱后添加一個(gè)簡(jiǎn)單的日志語(yǔ)句來(lái)驗(yàn)證這一點(diǎn)。

使用currentThread Scheduler：

Rx.Observable.from(arr).subscribe( ... );
console.log("Hi there!’);

"Total time: 8ms"
"Hi there!"

使用默認(rèn)Scheduler：

Rx.Observable.from(arr, null, null, Rx.Scheduler.timeout).subscribe( ... );
console.log("Hi there!’);

"Hi there!"
"Total time: 5423ms"

因?yàn)槭褂媚J(rèn)Schedule的Observer以異步方式發(fā)出其項(xiàng)目，所以我們的console.log語(yǔ)句（它是同步的）在Observable甚至開(kāi)始發(fā)出任何通知之前執(zhí)行。 使用currentThread Scheduler，所有通知都會(huì)同步發(fā)生，因此只有在Observable發(fā)出所有通知時(shí)才會(huì)執(zhí)行console.log語(yǔ)句。

因此，Scheduler確實(shí)可以改變我們的Observable的工作方式。 在我們的例子中，性能確實(shí)受到異步處理一個(gè)已經(jīng)可用的大型陣列的影響。 但我們實(shí)際上可以使用Scheduler來(lái)提高性能。 例如，我們可以在對(duì)Observable執(zhí)行昂貴的操作之前動(dòng)態(tài)切換Scheduler：

arr.groupBy(function(value) {
    return value % 2 === 0;
})
.map(function(value) {
    return value.observeOn(Rx.Scheduler.default);
})
.map(function(groupedObservable) {
    return expensiveOperation(groupedObservable);
});

在前面的代碼中，我們將數(shù)組中的所有值分為兩組：偶數(shù)和非偶數(shù)。 groupBy返回一個(gè)Observable，它為每個(gè)創(chuàng)建的組發(fā)出一個(gè)Observable。 這里是很酷的部分：在運(yùn)行之前對(duì)每個(gè)分組的Observable中的項(xiàng)目進(jìn)行昂貴的操作，我們使用observeOn將Scheduler切換到默認(rèn)值，這樣昂貴的操作將異步執(zhí)行，而不是阻塞事件循環(huán)

在上一節(jié)中，我們使用observeOn運(yùn)算符來(lái)更改某些Observable中的Scheduler。 observeOn和subscribeOn是返回Observable實(shí)例副本的運(yùn)算符，但它使用的Scheduler我們作為參數(shù)傳遞的。

observeOn接受一個(gè)Scheduler并返回一個(gè)使用該Scheduler的新Observable。 它將使每個(gè)onNext調(diào)用在新的Scheduler中運(yùn)行。

subscribeOn強(qiáng)制Observable的訂閱和取消訂閱工作（而不是通知）在特定的Scheduler上運(yùn)行。 與observeOn一樣，它接受Scheduler作為參數(shù)。 例如，當(dāng)我們?cè)跒g覽器中運(yùn)行并在訂閱調(diào)用中執(zhí)行重要工作時(shí)，卻不希望用它來(lái)阻止UI線程，subscribeOn非常有用。

讓我們?cè)谖覀儎倓偸褂玫腟cheduler中深入了解一下。 RxJS的運(yùn)算符最常用的是immediate，default和currentThread。

Immediate Scheduler同步發(fā)出來(lái)自O(shè)bservable的通知，因此無(wú)論何時(shí)在Immediate Scheduler上調(diào)度操作，它都將立即執(zhí)行，從而阻塞該線程。 Rx.Observable.range是內(nèi)部使用Immediate Scheduler序的運(yùn)算符之一：

console.log("Before subscription");

Rx.Observable.range(1, 5)
.do(function(a) {
    console.log("Processing value", a);
})
.map(function(value) { return value * value; })
.subscribe(function(value) { console.log("Emitted", value); });

console.log("After subscription");

Before subscription
Processing value 1
Emitted 1
Processing value 2
Emitted 4
Processing value 3
Emitted 9
Processing value 4
Emitted 16
Processing value 5
Emitted 25
After subscription

程序輸出按我們期望的順序發(fā)生。 每個(gè)console.log語(yǔ)句在當(dāng)前項(xiàng)的通知之前運(yùn)行。

Immediate Scheduler非常適合于在每個(gè)通知中執(zhí)行可預(yù)測(cè)且非常昂貴的操作的Observable。 此外，Observable最終必須調(diào)用onCompleted。

Default Scheduler以異步方式運(yùn)行操作。 您可以將其視為setTimeout的等價(jià)物，其延遲為零毫秒，從而保持序列中的順序。 它使用其運(yùn)行的平臺(tái)上可用的最有效的異步實(shí)現(xiàn)（例如，Node.js中的process.nextTick或?yàn)g覽器中的setTimeout）。

讓我們使用前一個(gè)使用了range示例，并使其在默認(rèn)的Scheduler上運(yùn)行。 為此，我們將使用observeOn運(yùn)算符：

console.log("Before subscription");

Rx.Observable.range(1, 5)
.do(function(value) {
    console.log("Processing value", value);
})
.observeOn(Rx.Scheduler.default)
.map(function(value) { return value * value; })
.subscribe(function(value) { console.log("Emitted", value); });

console.log("After subscription");

Before subscription
Processing value 1
Processing value 2
Processing value 3
Processing value 4
Processing value 5
After subscription
Emitted 1
Emitted 4
Emitted 9
Emitted 16
Emitted 25

這個(gè)輸出有很大的不同。 我們的同步console.log語(yǔ)句輸出每個(gè)值，但我們使Observable在默認(rèn)的Scheduler上運(yùn)行，它會(huì)異步生成每個(gè)值。 這意味著我們?cè)赿o運(yùn)算符中的日志語(yǔ)句在平方值之前處理。

Default Scheduler永遠(yuǎn)不會(huì)阻塞事件循環(huán)，因此它非常適合涉及時(shí)間的操作，如異步請(qǐng)求。 它也可以在從未完成的Observable中使用，因?yàn)樗粫?huì)在等待新通知時(shí)阻塞程序（這可能永遠(yuǎn)不會(huì)發(fā)生）。

currentThread Scheduler與Immediate Scheduler一樣是同步的，但是如果我們使用遞歸運(yùn)算符，它會(huì)將要執(zhí)行的操作排入隊(duì)列，而不是立即執(zhí)行它們。 遞歸運(yùn)算符是一個(gè)自己調(diào)度另一個(gè)運(yùn)算符的運(yùn)算符。 一個(gè)很好的例子就是repeat。 repeat運(yùn)算符 - 如果沒(méi)有給出參數(shù) - 將無(wú)限期地重復(fù)鏈中的先前Observable序列。

如果對(duì)使用Immediate Scheduler的運(yùn)算符（例如return）調(diào)用repeat，則會(huì)遇到麻煩。 讓我們通過(guò)重復(fù)值10來(lái)嘗試這個(gè)，然后使用take只取重復(fù)的第一個(gè)值。 理想情況下，代碼將打印10次然后退出：

// Be careful: the code below will freeze your environment!
Rx.Observable.return(10).repeat().take(1)
.subscribe(function(value) {
    console.log(value);
});

Error: Too much recursion

此代碼導(dǎo)致無(wú)限循環(huán)。 在訂閱時(shí)，如return調(diào)用onNext（10）然后onCompleted，這使得repeat再次訂閱return。 由于返回正在Immediate Scheduler上運(yùn)行，因此該過(guò)程會(huì)重復(fù)，導(dǎo)致無(wú)限循環(huán)并且永遠(yuǎn)不會(huì)結(jié)束。

但是如果相反我們通過(guò)將它作為第二個(gè)參數(shù)傳遞給currentThread Scheduler給return，我們得到：

var scheduler = Rx.Scheduler.currentThread;
Rx.Observable.return(10, scheduler).repeat().take(1)
.subscribe(function(value) {
    console.log(value);
});

10

現(xiàn)在，當(dāng)repeat重新訂閱返回時(shí)，新的onNext調(diào)用將排隊(duì)，因?yàn)橹暗膐nCompleted仍在發(fā)生。 repeat然后返回一個(gè)可以使用的一次性對(duì)象，它調(diào)用onCompleted并通過(guò)重復(fù)處理取消repeat，最終從subscribe返回調(diào)用。

currentThread Scheduler對(duì)于涉及遞歸運(yùn)算符（如repeat）的操作非常有用，并且通常用于包含嵌套運(yùn)算符的迭代。

對(duì)于諸如canvas或DOM動(dòng)畫(huà)之類(lèi)的快速視覺(jué)更新，我們可以使用具有非常小時(shí)間間隔的interval運(yùn)算符，或者我們可以在內(nèi)部使用類(lèi)似setTimeout的函數(shù)來(lái)調(diào)度通知。

但這兩種方法都不理想。 在他們兩個(gè)中我們都在瀏覽器上拋出所有這些更新，這可能無(wú)法足夠快地處理它們。之所以會(huì)發(fā)生這種情況，是因?yàn)闉g覽器正在嘗試渲染一個(gè)幀，然后它會(huì)收到渲染下一幀的指令，因此它會(huì)丟棄當(dāng)前幀以保持速度。 結(jié)果是導(dǎo)致動(dòng)畫(huà)的不流暢，卡頓。

瀏覽器具有處理動(dòng)畫(huà)的原生方式，并且它們提供了一個(gè)使用它的API，稱(chēng)為requestAnimationFrame。 requestAnimationFrame允許瀏覽器通過(guò)在最合適的時(shí)間排列動(dòng)畫(huà)來(lái)優(yōu)化性能，并幫助我們實(shí)現(xiàn)更流暢的動(dòng)畫(huà)。

RxDOM庫(kù)附帶了一些額外的調(diào)度程序，其中一個(gè)是requestAnimationFrame Scheduler。

是的，你猜對(duì)了。 我們可以使用此Scheduler來(lái)改進(jìn)我們的太空飛船視頻游戲。 在其中，我們建立了40ms的刷新速度 - 大約每秒25幀 - 通過(guò)在該速度下創(chuàng)建一個(gè)interval Observable，然后使用combineLatest以間隔設(shè)置的速度更新整個(gè)游戲場(chǎng)景（因?yàn)樗亲羁焖俑碌腛bservable） ）...但誰(shuí)知道瀏覽器使用這種技術(shù)丟幀了多少幀！ 使用requestAnimationFrame可以獲得更好的性能。

讓我們創(chuàng)建一個(gè)使用Rx.Scheduler.requestAnimationFrame作為其調(diào)度程序的Observable。 請(qǐng)注意，它與interval運(yùn)算符的工作方式類(lèi)似：

ch_schedulers/starfield_raf.js

function animationLoop(scheduler) {
    return Rx.Observable.generate(
        0,
        function() { return true; }, // Keep generating forever
        function(x) { return x + 1; }, // Increment internal value
        function(x) { return x; }, // Value to return on each notification
        Rx.Scheduler.requestAnimationFrame
    ); // Schedule to requestAnimationFrame
}

現(xiàn)在，無(wú)論何時(shí)我們使用了25 FPS動(dòng)畫(huà)，我們都可以使用animationLoop函數(shù)。 所以我們的Observable繪制了星星，之前看起來(lái)像這樣：

spaceship_reactive/spaceship.js

var StarStream = Rx.Observable.range(1, 250)
.map(function() {
    return {
        x: parseInt(Math.random() * canvas.width),
        y: parseInt(Math.random() * canvas.height),
        size: Math.random() * 3 + 1
    };
})
.toArray()
.flatMap(function(arr) {
    return Rx.Observable.interval(SPEED).map(function() {
        return arr.map(function(star) {
            if (star.y >= canvas.height) {
                star.y = 0;
            }
            star.y += 3;
            return star;
        });
    });
});

變成這樣：

ch_schedulers/starfield_raf.js

var StarStream = Rx.Observable.range(1, 250)
.map(function() {
    return {
        x: parseInt(Math.random() * canvas.width),
        y: parseInt(Math.random() * canvas.height),
        size: Math.random() * 3 + 1
    };
})
.toArray()
.flatMap(function(arr) {
    return animationLoop().map(function() {
        return arr.map(function(star) {
            if (star.y >= canvas.height) {
                star.y = 0;
            }
            star.y += 3;
            return star;
        });
    });
});

這給了我們一個(gè)更流暢的動(dòng)畫(huà)。 代碼也更簡(jiǎn)潔！

測(cè)試可能是我們可以使用Scheduler的最引人注目的場(chǎng)景之一。 到目前為止，在本書(shū)中，我們一直在編寫(xiě)我們的核心代碼而不考慮后果。 但是在現(xiàn)實(shí)世界的軟件項(xiàng)目中，我們將編寫(xiě)測(cè)試以確保我們的代碼按照我們的意圖運(yùn)行。

測(cè)試異步代碼很難。 我們經(jīng)常遇到以下問(wèn)題之一：

模擬異步事件很復(fù)雜且容易出錯(cuò)。 測(cè)試的重點(diǎn)是避免bug和錯(cuò)誤，但如果你的測(cè)試本身有錯(cuò)誤，那這顯然是有問(wèn)題的。

如果我們想要準(zhǔn)確測(cè)試基于時(shí)間的功能，自動(dòng)化測(cè)試變得非常緩慢。 例如，如果我們需要準(zhǔn)確測(cè)試在嘗試檢索遠(yuǎn)程文件四秒后調(diào)用錯(cuò)誤，則每個(gè)測(cè)試至少需要花費(fèi)很長(zhǎng)時(shí)間才能運(yùn)行結(jié)束。 如果我們不斷運(yùn)行我們的測(cè)試套件，那將影響我們的開(kāi)發(fā)時(shí)間。

RxJS為我們提供了TestScheduler，一個(gè)旨在幫助測(cè)試的Scheduler。 TestScheduler允許我們?cè)诜奖銜r(shí)模擬時(shí)間并創(chuàng)建確定性測(cè)試，確保它們100％可重復(fù)。 除此之外，它允許我們執(zhí)行需要花費(fèi)大量時(shí)間并將其壓縮到瞬間的操作，同時(shí)保持測(cè)試的準(zhǔn)確性。

TestScheduler是VirtualTimeScheduler的專(zhuān)業(yè)化。 VirtualTimeSchedulers在“虛擬”時(shí)間而不是實(shí)時(shí)執(zhí)行操作。 計(jì)劃的操作進(jìn)入隊(duì)列并在虛擬時(shí)間內(nèi)分配一個(gè)時(shí)刻。 然后，Scheduler在其時(shí)鐘前進(jìn)時(shí)按順序運(yùn)行操作。 因?yàn)樗翘摂M時(shí)間，所以一切都立即運(yùn)行，而不必等待指定的時(shí)間。 我們來(lái)看一個(gè)例子：

var onNext = Rx.ReactiveTest.onNext;
QUnit.test("Test value order", function(assert) {
    var scheduler = new Rx.TestScheduler();
    var subject = scheduler.createColdObservable(
        onNext(100, "first"),
        onNext(200, "second"),
        onNext(300, "third")
    );
    var result = "";
    subject.subscribe(function(value) { result = value });
    scheduler.advanceBy(100);
    assert.equal(result, "first");
    scheduler.advanceBy(100);
    assert.equal(result, "second");
    scheduler.advanceBy(100);
    assert.equal(result, "third");
});

在前面的代碼中，我們測(cè)試了來(lái)自冷Observable的一些值以正確的順序到達(dá)。 為此，我們?cè)赥estScheduler中使用helper方法createColdObservable來(lái)創(chuàng)建一個(gè)Observable，它回放我們作為參數(shù)傳遞的onNext通知。 在每個(gè)通知中，我們指定應(yīng)該發(fā)出通知值的時(shí)間。 在此之后，我們訂閱此Observable，手動(dòng)提前調(diào)度程序中的虛擬時(shí)間，并檢查它是否確實(shí)發(fā)出了預(yù)期值。 如果示例在正常時(shí)間運(yùn)行，則需要300毫秒，但由于我們使用TestScheduler來(lái)運(yùn)行Observable，它將立即運(yùn)行，但完全按照我們的順序。

沒(méi)有比在現(xiàn)實(shí)世界中為時(shí)間敏感的任務(wù)編寫(xiě)測(cè)試更好的方法來(lái)理解如何使用虛擬時(shí)間來(lái)縮短時(shí)間。 讓我們從我們?cè)诰彌_值中制作的地震查看器中恢復(fù)一個(gè)Observable：

quakes
.pluck("properties")
.map(makeRow)
.bufferWithTime(500)
.filter(function(rows) { return rows.length > 0; })
.map(function(rows) {
    var fragment = document.createDocumentFragment();
    rows.forEach(function(row) {
        fragment.appendChild(row);
    });
    return fragment;
})
.subscribe(function(fragment) {
    table.appendChild(fragment);
});

為了使代碼更易于測(cè)試，讓我們將Observable封裝在一個(gè)函數(shù)中，該函數(shù)接受我們?cè)赽ufferWithTime運(yùn)算符中使用的Scheduler。在Obpectables中參數(shù)化將要測(cè)試的Scheduler總是一個(gè)好主意。

ch_schedulers/testscheduler.js

function quakeBatches(scheduler) {
    return quakes.pluck("properties")
    .bufferWithTime(500, null, scheduler || null)
    .filter(function(rows) {
        return rows.length > 0;
    });
}

讓我們通過(guò)采取一些步驟來(lái)簡(jiǎn)化代碼，但保持本質(zhì)。 此代碼采用包含屬性屬性的Observable JSON對(duì)象，將它們緩沖到每500毫秒釋放的批次中，并過(guò)濾掉空的批次。

我們想要驗(yàn)證此代碼是否有效，但我們絕對(duì)不希望每次運(yùn)行測(cè)試時(shí)都等待幾秒鐘，以確保我們的緩沖按預(yù)期工作。 這是虛擬時(shí)間和TestScheduler將幫助我們的地方：

ch_schedulers/testscheduler.js

? var onNext = Rx.ReactiveTest.onNext;
var onCompleted = Rx.ReactiveTest.onCompleted;
var subscribe = Rx.ReactiveTest.subscribe;
? var scheduler = new Rx.TestScheduler();
? var quakes = scheduler.createHotObservable(
    onNext(100, { properties: 1 }),
    onNext(300, { properties: 2 }),
    onNext(550, { properties: 3 }),
    onNext(750, { properties: 4 }),
    onNext(1000, { properties: 5 }),
    onCompleted(1100)
);
? QUnit.test("Test quake buffering", function(assert) {
    ? var results = scheduler.startScheduler(function() {
        return quakeBatches(scheduler)
    }, {
        created: 0,
        subscribed: 0,
        disposed: 1200
    });
    ? var messages = results.messages;
    console.log(results.scheduler === scheduler);
    ? assert.equal(
        messages[0].toString(),
        onNext(501, [1, 2]).toString()
    );
    assert.equal(
        messages[1].toString(),
        onNext(1001, [3, 4, 5]).toString()
    );
    assert.equal(
        messages[2].toString(),
        onCompleted(1100).toString()
    );
});

讓我們一步一步地剖析代碼：

我們首先從ReactiveTest加載一些輔助函數(shù)。 這些在虛擬時(shí)間內(nèi)注冊(cè)onNext，onCompleted和訂閱事件。

我們創(chuàng)建了一個(gè)新的TestScheduler，它將推動(dòng)整個(gè)測(cè)試。

我們使用TestScheduler中的方法createHotObservable創(chuàng)建一個(gè)假的熱Observable，它將在虛擬時(shí)間內(nèi)模擬特定點(diǎn)的通知。 特別是，它在第一秒發(fā)出五個(gè)通知，并在1100毫秒完成。 每次它發(fā)出一個(gè)具有特定屬性的對(duì)象。

我們可以使用任何測(cè)試框架來(lái)運(yùn)行測(cè)試。 對(duì)于我們的例子，我選擇了QUnit。

我們使用startScheduler方法創(chuàng)建一個(gè)使用測(cè)試調(diào)度程序的Observable。 第一個(gè)參數(shù)是一個(gè)函數(shù)，它創(chuàng)建Observable以使用我們的Scheduler運(yùn)行。 在我們的例子中，我們只返回我們傳遞TestScheduler的quakeBatches函數(shù)。 第二個(gè)參數(shù)是一個(gè)對(duì)象，它包含我們想要?jiǎng)?chuàng)建Observable的不同虛擬時(shí)間，訂閱它并處理它。 對(duì)于我們的示例，我們?cè)谔摂M時(shí)間0開(kāi)始和訂閱，并且我們?cè)?200（虛擬）毫秒處理Observable。

startScheduler方法返回一個(gè)帶有scheduler和messages屬性的對(duì)象。 在消息中，我們可以在虛擬時(shí)間內(nèi)找到Observable發(fā)出的所有通知。

我們的第一個(gè)斷言測(cè)試在501毫秒之后（在第一個(gè)緩沖時(shí)間限制之后），我們的Observable產(chǎn)生值1和2。
我們的第二個(gè)斷言測(cè)試在1001毫秒后，我們的Observable產(chǎn)生剩余的值3,4和5.最后，我們的第三個(gè)斷言檢查序列是否完全在1100毫秒完成，正如我們?cè)跓岬腛bservable地震中所指出的那樣。

該代碼以非常可靠的方式有效地測(cè)試我們的高度異步的Observable，并且無(wú)需跳過(guò)箍來(lái)模擬異步條件。我們只是指定我們希望代碼在虛擬時(shí)間內(nèi)作出反應(yīng)的時(shí)間，我們使用測(cè)試調(diào)度程序來(lái)運(yùn)行整個(gè)操作。

Scheduler是RxJS的重要組成部分。 即使您可以在沒(méi)有明確使用它們的情況下走很長(zhǎng)的路，它們也是一種先進(jìn)的概念，它可以讓您在程序中微調(diào)并發(fā)性。虛擬時(shí)間的概念是RxJS獨(dú)有的，對(duì)于測(cè)試異步代碼等任務(wù)非常有用。

在下一章中，我們將使用Cycle.js，這是一種基于稱(chēng)為單向數(shù)據(jù)流的概念來(lái)創(chuàng)建令人驚嘆的Web應(yīng)用程序的反應(yīng)方式。有了它，我們將使用現(xiàn)代技術(shù)創(chuàng)建一個(gè)快速的Web應(yīng)用程序，從而顯著改進(jìn)傳統(tǒng)的Web應(yīng)用程序制作方式。

關(guān)注我的微信公眾號(hào)，更多優(yōu)質(zhì)文章定時(shí)推送