后端【免費(fèi)下載鏈接】RxJSThe Reactive Extensions for JavaScript項(xiàng)目地址https://gitcode.com/gh_mirrors/rxj/RxJS點(diǎn)擊查看免費(fèi)下載導(dǎo)讀Rx.Observable.prototype.takeLastWithTime(duration)是 RxJS v4Reactive Extensions for JavaScript中一個(gè)按時(shí)間維度截取序列末尾元素的時(shí)間類運(yùn)算符它在源序列結(jié)束后僅把「結(jié)束時(shí)刻往前推duration毫秒窗口內(nèi)」產(chǎn)生的元素發(fā)射給下游同時(shí)用指定的調(diào)度器Scheduler負(fù)責(zé)計(jì)時(shí)與緩沖隊(duì)列的排空。本指南以 官方 API 文檔 為主線結(jié)合倉(cāng)庫(kù)中的核心實(shí)現(xiàn)src/core/linq/observable/takelastwithtime.js、模塊化實(shí)現(xiàn)src/modular/observable/takelastwithtime.js與兩套單元測(cè)試講解其調(diào)用簽名、默認(rèn)調(diào)度器行為、內(nèi)部滑動(dòng)緩沖隊(duì)列原理、邊界條件以及完整的可運(yùn)行示例。讀完本文你將能熟練使用takeLastWithTime實(shí)現(xiàn)「只看最近一段時(shí)間產(chǎn)生的數(shù)據(jù)」這類時(shí)間窗口需求并能讀懂其源碼級(jí)的時(shí)間判定邏輯。一、功能概述takeLastWithTime的官方語(yǔ)義是Returns elements within the specified duration from the end of the observable source sequence, using the specified schedulers to run timers and to drain the collected elements.即返回源序列末尾指定時(shí)間范圍內(nèi)產(chǎn)生的元素。它只關(guān)心「時(shí)間」不關(guān)心「?jìng)€(gè)數(shù)」——與按個(gè)數(shù)截取末尾元素的takeLast系列形成對(duì)比。源序列一旦發(fā)出onCompleted通知運(yùn)算符就會(huì)以此刻為基準(zhǔn)從內(nèi)部隊(duì)列中取出「結(jié)束時(shí)刻 ? duration ≤ 元素產(chǎn)生時(shí)刻 ≤ 結(jié)束時(shí)刻」的所有元素依次發(fā)射后完成。與它互補(bǔ)的姊妹運(yùn)算符是 skipLastWithTime后者跳過末尾時(shí)間窗口內(nèi)的元素、只發(fā)射窗口之外的舊元素二者內(nèi)部都使用「帶時(shí)間戳的滑動(dòng)隊(duì)列」這一相同的數(shù)據(jù)結(jié)構(gòu)。二、方法簽名與參數(shù)說明Rx.Observable.prototype.takeLastWithTime(duration, [timeScheduler], [loopScheduler])官方文檔給出的參數(shù)定義如下參數(shù)類型說明默認(rèn)值durationNumber從序列末尾開始取元素的持續(xù)時(shí)間毫秒即時(shí)間窗口寬度必填timeSchedulerScheduler負(fù)責(zé)運(yùn)行定時(shí)器的調(diào)度器Rx.Scheduler.timeoutloopSchedulerScheduler負(fù)責(zé)排空drain已收集元素的調(diào)度器Rx.Scheduler.currentThread返回值Observable—— 一個(gè)包含源序列末尾指定時(shí)長(zhǎng)內(nèi)元素的 observable 序列。一個(gè)值得注意的文檔與實(shí)現(xiàn)差異需要指出的是當(dāng)前倉(cāng)庫(kù)的實(shí)際實(shí)現(xiàn)只接收一個(gè)調(diào)度器參數(shù)。核心實(shí)現(xiàn)位于 src/core/linq/observable/takelastwithtime.jsobservableProto.takeLastWithTime function (duration, scheduler) { isScheduler(scheduler) || (scheduler defaultScheduler); return new TakeLastWithTimeObservable(this, duration, scheduler); };其 JSDoc 注釋同樣只聲明了[scheduler]一個(gè)可選參數(shù)默認(rèn)值為Rx.Scheduler.timeout。模塊化版本src/modular/observable/takelastwithtime.js也保持一致只是在默認(rèn)值上使用Scheduler.asyncmodule.exports function takeLastWithTime (source, duration, scheduler) { Scheduler.isScheduler(scheduler) || (scheduler Scheduler.async); return new TakeLastWithTimeObservable(source, duration, scheduler); };而Rx.Scheduler.timeout與Rx.Scheduler.async實(shí)為同一調(diào)度器。從 src/core/concurrency/defaultscheduler.js 可以看到它們的綁定關(guān)系var defaultScheduler Scheduler[default] Scheduler.async new DefaultScheduler();因此實(shí)踐中直接使用takeLastWithTime(duration)或takeLastWithTime(duration, scheduler)即可文檔中列出的第二個(gè)loopScheduler參數(shù)在當(dāng)前實(shí)現(xiàn)中并不生效它以單一scheduler同時(shí)承擔(dān)「取時(shí)間戳」「驅(qū)動(dòng)滑動(dòng)窗口」與「排空隊(duì)列」的職責(zé)。三、底層實(shí)現(xiàn)原理帶時(shí)間戳的滑動(dòng)緩沖隊(duì)列takeLastWithTime的完整實(shí)現(xiàn)由兩個(gè)類協(xié)作完成全部代碼位于 src/core/linq/observable/takelastwithtime.js。3.1 外層 Observable訂閱時(shí)掛載觀察者var TakeLastWithTimeObservable (function (__super__) { inherits(TakeLastWithTimeObservable, __super__); function TakeLastWithTimeObservable(source, d, s) { this.source source; this._d d; this._s s; __super__.call(this); } TakeLastWithTimeObservable.prototype.subscribeCore function (o) { return this.source.subscribe(new TakeLastWithTimeObserver(o, this._d, this._s)); }; // ... }(ObservableBase));它繼承自O(shè)bservableBase只做一件事訂閱發(fā)生時(shí)把下游觀察者o、時(shí)長(zhǎng)_d和調(diào)度器_s一并封裝進(jìn)內(nèi)部觀察者再訂閱源序列。3.2 核心觀察者next階段維護(hù)滑動(dòng)窗口var TakeLastWithTimeObserver (function (__super__) { inherits(TakeLastWithTimeObserver, __super__); function TakeLastWithTimeObserver(o, d, s) { this._o o; this._d d; this._s s; this._q []; __super__.call(this); } TakeLastWithTimeObserver.prototype.next function (x) { var now this._s.now(); this._q.push({ interval: now, value: x }); while (this._q.length 0 now - this._q[0].interval this._d) { this._q.shift(); } }; // ... }(AbstractObserver));這里體現(xiàn)了整個(gè)運(yùn)算符最關(guān)鍵的設(shè)計(jì)帶時(shí)間戳入隊(duì)每個(gè)到達(dá)的元素x不是直接進(jìn)隊(duì)列而是以調(diào)度器時(shí)鐘this._s.now()打上時(shí)間戳包裝成{ interval: now, value: x }對(duì)象。因此時(shí)間基準(zhǔn)完全由調(diào)度器決定——換用TestScheduler即可在虛擬時(shí)間下進(jìn)行確定性測(cè)試。隊(duì)首過期即淘汰每次入隊(duì)后只要隊(duì)首元素的時(shí)間戳距今已超過duration即now - 隊(duì)首.interval duration就從隊(duì)首彈出。這保證隊(duì)列中永遠(yuǎn)只保留「最近duration毫秒內(nèi)」到達(dá)的元素是一個(gè)真正的滑動(dòng)窗口而不是等到完成時(shí)才一次性篩選。隊(duì)尾單調(diào)性假設(shè)淘汰邏輯只檢查隊(duì)首隱含假設(shè)元素按時(shí)間單調(diào)到達(dá)對(duì)亂序源序列takeLastWithTime并不保證按真實(shí)到達(dá)時(shí)間重新排序。3.3 完成階段按窗口放行元素TakeLastWithTimeObserver.prototype.completed function () { var now this._s.now(); while (this._q.length 0) { var next this._q.shift(); if (now - next.interval this._d) { this._o.onNext(next.value); } } this._o.onCompleted(); };onCompleted到達(dá)時(shí)以當(dāng)前時(shí)刻為基準(zhǔn)再次篩選隊(duì)列中所有「結(jié)束時(shí)刻 ? 產(chǎn)生時(shí)刻 ≤ duration」的元素按入隊(duì)順序即到達(dá)順序依次onNext發(fā)射最后onCompleted。注意此處的判定是包含窗口邊界而next中淘汰過期元素用的是兩處邊界語(yǔ)義正好互補(bǔ)避免邊界元素被誤刪。錯(cuò)誤路徑則直接透?jìng)鱡rror不做任何緩沖立即onError(e)轉(zhuǎn)發(fā)給下游。3.4 三階段行為總覽源事件運(yùn)算符行為下游觀察onNext(x)打時(shí)間戳入隊(duì)彈出隊(duì)首超過duration的舊元素不發(fā)射任何元素onError(e)清空隊(duì)列語(yǔ)義不適用直接透?jìng)鱫nError(e)onCompleted()以當(dāng)前時(shí)刻為界發(fā)射窗口內(nèi)全部元素依次onNext后onCompleted四、完整可運(yùn)行示例官方示例官方文檔給出了如下可直接在瀏覽器控制臺(tái)或 Node 中運(yùn)行的示例完整代碼見 doc/api/core/operators/takelastwithtime.mdvar source Rx.Observable.timer(0, 1000) .take(10) .takeLastWithTime(5000); var subscription source.subscribe( function (x) { console.log(Next: x); }, function (err) { console.log(Error: err); }, function () { console.log(Completed); }); // Next: 5 // Next: 6 // Next: 7 // Next: 8 // Next: 9 // Completed運(yùn)行分析Rx.Observable.timer(0, 1000).take(10)產(chǎn)生元素0, 1, 2, …, 9分別在第 0、1、2、…、9 秒發(fā)出第 9 秒全部發(fā)完后onCompletedtakeLastWithTime(5000)要求只保留「完成時(shí)刻前 5 秒內(nèi)」的元素。完成發(fā)生在第 9 秒窗口起點(diǎn)為第 4 秒因此0–4全部被淘汰5–9全部保留并依次輸出輸出順序與源序列到達(dá)順序一致5, 6, 7, 8, 9隨后Completed。五、引入方式與發(fā)布產(chǎn)物該運(yùn)算符隨 RxJS v4 的多個(gè)構(gòu)建產(chǎn)物一起發(fā)布。官方文檔列出的分發(fā)文件與引用關(guān)系如下對(duì)應(yīng)本倉(cāng)庫(kù)modules/目錄下的實(shí)際產(chǎn)物全量構(gòu)建modules/rx-core/rx.core.js 之外的rx.all.js、rx.all.compat.js時(shí)間運(yùn)算符集合rx.time.js對(duì)應(yīng) modules/rx-lite-time/rx.lite.time.js 等同名產(chǎn)物精簡(jiǎn)版rx.lite.js、rx.lite.compat.js對(duì)應(yīng) modules/rx-lite/rx.lite.js前置依賴若單獨(dú)使用rx.time.js必須先加載核心rx.js/rx.compat.js因?yàn)閠akeLastWithTime依賴ObservableBase、AbstractObserver等基礎(chǔ)設(shè)施包管理NPM 包為rxNuGet 包為RxJS-All、RxJS-Time、RxJS-Lite對(duì)應(yīng)倉(cāng)庫(kù) nuget 目錄中的RxJS-All.nuspec、RxJS-Time.nuspec、RxJS-Lite.nuspec。在瀏覽器中按順序引入核心與時(shí)間模塊后即可通過Rx.Observable.prototype.takeLastWithTime調(diào)用。六、單元測(cè)試與邊界行為驗(yàn)證倉(cāng)庫(kù)提供了兩套測(cè)試經(jīng)典 QUnit 版位于 tests/observable/takelastwithtime.js模塊化 tape 版位于 src/modular/test/takelastwithtime.js二者用例結(jié)構(gòu)一致均用TestScheduler在虛擬時(shí)間下驗(yàn)證行為。以下為 QUnit 版的關(guān)鍵用例歸納測(cè)試名輸入虛擬時(shí)間duration期望結(jié)果zero 1/zero 2210/220/230 產(chǎn)生元素230 完成0只onCompleted(230)無(wú)任何元素some 1210→1220→2230→3240 完成25onNext(240,2)、onNext(240,3)、onCompleted(240)some 2210→1220→2230→3300 完成25僅onCompleted(300)元素全部過期some 3210–290 每 10ms 一個(gè)元素300 完成45onNext(300,6..9)、onCompleted(300)some 4210–300 稀疏到達(dá)350 完成25僅onCompleted(350)元素全部過期all210→1220→2230 完成50onNext(230,1)、onNext(230,2)、onCompleted(230)error210 拋錯(cuò)50onError(210, error)無(wú)緩沖直接透?jìng)鱪ever永不完成50無(wú)任何消息訂閱持續(xù)到 1000從這些用例可以提煉出四個(gè)關(guān)鍵邊界結(jié)論duration 0時(shí)結(jié)果為空now - interval 0使所有已入隊(duì)元素立即被淘汰測(cè)試zero 1/2元素在完成瞬間是否保留取決于「完成時(shí)刻 ? 產(chǎn)生時(shí)刻」是否 duration測(cè)試some 1中 220 時(shí)刻的元素被保留而 210 時(shí)刻的元素被淘汰邊界精確到虛擬時(shí)間刻度窗口完全落在過去時(shí)序列「靜默完成」一個(gè)元素都不發(fā)射僅產(chǎn)生onCompleted測(cè)試some 2、some 4錯(cuò)誤不參與緩沖源序列報(bào)錯(cuò)時(shí)立即轉(zhuǎn)發(fā)隊(duì)列中已收集的元素被丟棄測(cè)試error。七、典型應(yīng)用場(chǎng)景結(jié)合上述行為特征takeLastWithTime適合以下場(chǎng)景「只看最近 N 秒數(shù)據(jù)」的儀表盤/監(jiān)控面板訂閱一個(gè)持續(xù)產(chǎn)生事件的熱序列事件結(jié)束后只關(guān)心收尾階段最近幾秒的指標(biāo)忽略較早的歷史數(shù)據(jù)回放與審計(jì)在流結(jié)束時(shí)只取出「最近一段時(shí)間內(nèi)發(fā)生的事件」用于審計(jì)或日志裁剪與TestScheduler結(jié)合做確定性時(shí)間測(cè)試由于時(shí)間戳全部取自注入的調(diào)度器可以在虛擬時(shí)間里精確斷言「哪些元素落在結(jié)束時(shí)刻前duration窗口內(nèi)」按時(shí)間而非個(gè)數(shù)截取末尾當(dāng)你不關(guān)心元素個(gè)數(shù)、只關(guān)心時(shí)間跨度時(shí)例如「只保留最后 5 秒內(nèi)的采樣點(diǎn)」它比按個(gè)數(shù)取末尾更貼合需求。需要提醒的是它是一個(gè)被動(dòng)緩存型運(yùn)算符——在源序列完成之前它不會(huì)向下游發(fā)射任何元素所有候選元素都會(huì)先進(jìn)入內(nèi)部隊(duì)列因此不適合需要實(shí)時(shí)輸出的場(chǎng)景如果你希望「實(shí)時(shí)發(fā)射且丟棄過早的元素」應(yīng)改用時(shí)間滑窗類運(yùn)算符如bufferWithTime、windowWithTime等見 doc/api/core/operators 目錄下的相關(guān)文檔。八、相關(guān)資源官方 API 文檔doc/api/core/operators/takelastwithtime.md核心實(shí)現(xiàn)src/core/linq/observable/takelastwithtime.js模塊化實(shí)現(xiàn)src/modular/observable/takelastwithtime.js單元測(cè)試QUnittests/observable/takelastwithtime.js單元測(cè)試tapesrc/modular/test/takelastwithtime.js互補(bǔ)運(yùn)算符skipLastWithTime 文檔默認(rèn)調(diào)度器定義src/core/concurrency/defaultscheduler.js贊分享后端【免費(fèi)下載鏈接】RxJSThe Reactive Extensions for JavaScript項(xiàng)目地址https://gitcode.com/gh_mirrors/rxj/RxJS點(diǎn)擊查看免費(fèi)下載相關(guān)推薦RxJS 4 skipLastWithTime 操作符深度解析按時(shí)間窗口跳過序列末尾元素RxJS 4 skipLastWithTime 操作符深度解析按時(shí)間窗口跳過序列末尾元素 skipLastWithTime duration, schedul后端RxJS 4 運(yùn)算符詳解takeLast(count) 從序列末尾截取元素的緩沖式實(shí)現(xiàn)原理與實(shí)戰(zhàn)RxJS 4 運(yùn)算符詳解takeLast count 從序列末尾截取元素的緩沖式實(shí)現(xiàn)原理與實(shí)戰(zhàn) 本指南以 RxJS v4Reactive Extension后端RxJS 4 takeLastBuffer 操作符深入解析從序列末尾一次性提取指定數(shù)量元素的數(shù)組RxJS 4 takeLastBuffer 操作符深入解析從序列末尾一次性提取指定數(shù)量元素的數(shù)組 takeLastBuffer 是 RxJS v4 中一個(gè)看似后端上一篇AutoBangumi API文檔自動(dòng)生成FastAPI與Swagger整合下一篇為什么SiYuan的塊級(jí)知識(shí)管理比傳統(tǒng)筆記軟件更高效創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考