RxJS 4 的 `Rx.Observable.repeat` 深入解析：固定值重复发射、调度器与源码实现-北京尧图网络

RxJS 4 的Rx.Observable.repeat深入解析固定值重复发射、调度器与源码实现【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址: https://gitcode.com/gh_mirrors/rxj/RxJS导读Rx.Observable.repeat是 RxJS 4Reactive Extensions for JavaScript中一个非常基础但高频使用的静态工厂方法它生成一个可观察序列Observable把给定的同一个元素按指定次数重复发射出去。本文以 doc/api/core/operators/repeat.md 为核心完整讲解它的参数语义、默认行为、无限重复与退订机制并结合仓库中的性能优化实现、可枚举序列实现与全套单元测试从 API 用法一路深入到scheduleRecursive递归调度的底层原理帮助你既会用、又看得懂它的内部机制。一、API 签名与核心语义repeat的完整签名如下Rx.Observable.repeat(value, [repeatCount], [scheduler])它生成一个可观察序列该序列使用指定的调度器scheduler发送观察者消息将给定的元素重复指定次数。换言之它产出的序列只包含value这一个值区别只在于次数有限发射repeatCount次后正常完成onCompleted次数无限不指定repeatCount时无限重复发射永不完成。一个直观的对比是它与Rx.Observable.return即just的关系return(value)只发射一次repeat(value, n)相当于把return的序列自我复制了 n 次再串联起来。二、参数详解与默认值参数类型必填默认值说明valueAny是无要重复发射的元素可以是任意 JS 值数字、字符串、对象等repeatCountNumber否-1重复发射的次数不传或传null时等价于无限重复schedulerScheduler否Scheduler.immediate运行生产者循环的调度器关于默认调度器文档标注为Scheduler.immediate同步执行值得注意的是本仓库的性能优化版本 src/core/perf/operators/repeat.js 中实际写法是Observable.repeat function (value, repeatCount, scheduler) { isScheduler(scheduler) || (scheduler currentThreadScheduler); return new RepeatObservable(value, repeatCount, scheduler); };即在不传调度器时回退到currentThreadScheduler。无论immediate还是currentThreadScheduler它们都属于同步执行调度器对绝大多数调用方来说行为一致只有在希望把发射循环推迟到某个队列或异步调度器上执行时才需要显式传入第三个参数。关于repeatCount的边界语义可从 tests/observable/repeat.js 的测试中逐一确认repeat(42, 0)一个值都不发射直接onCompletedrepeat(42, 1)发射 1 次42然后onCompletedrepeat(42, 10)发射 10 次42然后onCompletedrepeat(42, -1)/repeat(42)无限发射只有显式退订才会停止。三、快速上手完整的可运行示例原文示例直接可运行这里保留完整形态并补充注释var source Rx.Observable.repeat(42, 3); var subscription source.subscribe( function (x) { console.log(Next: x); }, function (err) { console.log(Error: err); }, function () { console.log(Completed); }); // Next: 42 // Next: 42 // Next: 42 // Completed输出结果与参数严格对应repeatCount 3时控制台打印三个Next: 42随后触发Completed。若把第二参数去掉改为Rx.Observable.repeat(42)则Next: 42会无限打印下去永远不会走到Completed分支——这正是无限重复的真实运行效果。四、无限重复与退订Dispose无限序列在浏览器控制台或 Node 脚本中运行会一直占用主线程输出。正确的做法是通过订阅返回的subscription主动退订var source Rx.Observable.repeat(42); // 不传次数无限 var count 0; var subscription source.subscribe(function (x) { console.log(Next: x); if (count 5) { subscription.dispose(); // 手动停止 } }); // Next: 42 共 5 次后停止这一点在单元测试中有专门覆盖tests/observable/repeat.js的repeat value count dispose与repeat value两个用例都是通过TestScheduler在指定虚拟时刻如disposed: 207终止订阅从而验证有限次数未跑完时退订能立即中断以及无限重复可通过退订中断。五、与原型方法observableProto.repeat的区别除了静态工厂Rx.Observable.repeat(value, ...)RxJS 4 还提供了对已有序列重复的实例方法var source Rx.Observable.return(1).repeat(3);它的实现位于 src/core/linq/observable/repeatproto.jsobservableProto.repeat function (repeatCount) { return enumerableRepeat(this, repeatCount).concat(); };这里复用了 src/core/enumerable.js 中的RepeatEnumerable/RepeatEnumerator把源序列本身作为可枚举对象用concat()将同一序列反复串联。注意两者的语义差异静态Observable.repeat(value, n)重复的是单个值原型source.repeat(n)重复的是整个序列——源序列每完成一次就重新订阅并重放一遍直到凑满 n 轮或无限轮。对应地retry操作符在 src/core/linq/observable/retry.js 中就是enumerableRepeat(this, retryCount).catchError()可见repeat与重试是同一套枚举-串联机制的两种用法。六、源码级剖析RepeatObservable与递归调度性能优化版实现在 src/core/perf/operators/repeat.js核心是两个对象1.RepeatObservable继承ObservableBasefunction RepeatObservable(value, repeatCount, scheduler) { this.value value; this.repeatCount repeatCount null ? -1 : repeatCount; // null → -1 → 无限 this.scheduler scheduler; __super__.call(this); }构造函数在这里完成了默认值归一化repeatCount null一律转为-1后续循环逻辑以-1作为无限哨兵值。2.RepeatSink的run方法发射循环本体RepeatSink.prototype.run function () { var observer this.observer, value this.parent.value; function loopRecursive(i, recurse) { if (i -1 || i 0) { observer.onNext(value); i 0 i--; } if (i 0) { return observer.onCompleted(); } recurse(i); } return this.parent.scheduler.scheduleRecursive(this.parent.repeatCount, loopRecursive); };这段代码是整个操作符的灵魂逐行解读循环携带状态i剩余发射次数初值为repeatCounti -1无限或i 0还有剩余时发射一次value有限模式下同时自减i 0时调用observer.onCompleted()终止整个序列否则通过回调参数recurse(i)触发下一轮。scheduleRecursive来自 src/core/concurrency/scheduler.recursive.js其实现为schedulerProto.scheduleRecursive function (state, action) { return this.schedule([state, action], invokeRecImmediate); };它把状态动作打包成一次调度由invokeRecImmediate在动作内部循环调用recurse从而做到不爆栈的递归式循环发射。这也是RepeatSink不需要显式持有订阅对象却能响应退订的原因调度器返回的 disposable 就是整个循环的取消句柄。此外模块化版本 src/modular/observable/repeat.js 给出了另一种等价实现它把value包装成一个符合iterator协议、带remaining计数器的迭代器repeatValue再用ConcatObservablescheduleRecursive逐个消费迭代器项并转发给下游观察者。这种迭代器串联的结构与observableProto.repeat的实现思路一脉相承可以对照阅读。七、单元测试验证行为契约一览完整的测试用例位于 tests/observable/repeat.js覆盖了静态工厂与原型方法的全部关键分支是理解行为契约的最佳参考测试组验证点repeat value count zero / one / ten有限次数下发射个数精确结束后onCompletedrepeat value count dispose有限次数未跑完时退订立即中断repeat valuerepeatCount -1即无限重复靠退订停止Repeat Observable basic / infinite原型方法对冷序列重放源不完成则只订阅一次Repeat Observable error源序列出错时错误透传、不再重试Repeat Observable throws观察者回调抛异常、源抛异常时正确上抛Repeat Observable repeat count Basic / dispose / infinite / error / throws带次数的原型方法全部边界例如repeat value count ten断言了从虚拟时刻 201 到 210 连续 10 个onNext(20x, 42)再于 210 时刻onCompletedRepeat Observable basic则用xs.subscriptions.assertEqual精确断言了冷序列被反复订阅的时间窗subscribe(200, 450)、subscribe(450, 700)……直观展示了序列重复重新订阅的本质。同目录下还有模块化测试 src/modular/test/repeat.js 与性能测试 tests/perf/operators/repeat.js 可供对照。八、使用建议与注意事项明确区分重复值与重复序列需要单值重放用Rx.Observable.repeat(value, n)需要整段序列循环用source.repeat(n)无限重复务必可控不传repeatCount的序列永不onCompleted一定要保留订阅句柄并在适当时候dispose()否则会造成持续输出甚至资源占用调度器按需显式指定默认的同步调度器意味着repeat会在调用线程上立即倾泻全部消息若要在定时器、队列或虚拟时间测试环境下运行请显式传入scheduler例如Rx.Observable.repeat(42, 3, Rx.Scheduler.default)与重试机制的亲缘关系retry在底层就是repeat与catchError的组合理解repeat的枚举-串联实现等于同时理解了重试操作符的骨架。九、配套资源定位API 文档doc/api/core/operators/repeat.md性能优化实现src/core/perf/operators/repeat.js原型方法实现src/core/linq/observable/repeatproto.js可枚举序列RepeatEnumerablesrc/core/enumerable.js递归调度器src/core/concurrency/scheduler.recursive.js模块化实现src/modular/observable/repeat.js单元测试tests/observable/repeat.js、src/modular/test/repeat.js、tests/perf/operators/repeat.js相关操作符retrysrc/core/linq/observable/retry.js【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址: https://gitcode.com/gh_mirrors/rxj/RxJS创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

RxJS 4 的 `Rx.Observable.repeat` 深入解析：固定值重复发射、调度器与源码实现

相关推荐

FreshRSS 用户管理实战指南：账号创建、自注册开关、邮箱验证与 ToS 配置

swagger-codegen TypeScript-Aurelia 客户端生成器完全指南：从 OpenAPI 定义到可注入 Aurelia 的 NPM 模块

Ory Hydra JsonWebKey 模型完全指南：字段语义、SDK 方法与 Admin API 密钥管理实战

更多资讯

Java 21 + Spring Boot 3 构建企业级 RAG 智能体工作流引擎

system.img修改全攻略：格式识别、解包重打包与刷写避坑指南

变频与定频空调负荷聚合模型及Matlab实现

MyBatis与EHCache整合优化数据库性能实践

Spring AI实战：Java开发者快速集成AI能力指南

Chrome Extensions Samples 之 Multi-Tab Browser：基于 webview 的 New Window、User Agent 与权限请求综合示例解析

今日更新

OneUptime 自定义探针（Custom Probe）部署实战：私网监控、代理配置与断连排障全指南

大众TL52625前端框架材料要求详解：从性能测试到落地执行

TiXL 浮点运算算子库 Lib.numbers.float 完全指南：44 个算子的参数详解、源码原理与实战串联

本周精选

深入解析Transformer多头注意力机制与工程优化

OpenClaw 的 Skills 跑学习任务，模型通道改到 TaoToken 通道行不行？

ChatGPT报错Oops, an error occurred! 全链路排查指南

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？