RxJS v4 操作符分类指南按 11 大类掌握 Observable 的创建、转换、组合与调度【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址: https://gitcode.com/gh_mirrors/rxj/RxJS本文是 RxJS v4The Reactive Extensions for JavaScript操作符体系的分类总索引基于仓库 doc/gettingstarted/categories.md 展开将Observable类型实现的全部主要操作符划分为创建、转换、组合、函数式、数学、时间、异常、筛选、分组、命令式与原语等类别。读完本文你将获得一张完整的操作符地图知道每个操作符属于哪一类、解决什么问题、以静态方法还是实例方法调用并可通过文中指向的 操作符 API 文档 与源码文件继续深入每个操作符的签名、参数与测试用例。文档定位与阅读方式本文对应的原始文档是操作符总目录它列出了由Observable类型实现的所有主要操作符并按用途归入 11 个类别。在 RxJS v4 中操作符有两种挂载形态静态方法挂在Rx.Observable上用于创建序列或从外部资源转换出序列例如Rx.Observable.range(0, 5)实例方法prototype挂在Observable.prototype上对已有序列进行变换、组合、筛选等例如source.map(x x * 2)。同一功能在分类表中同时出现静态与实例两个条目时如amb与prototype.amb前者传入多个序列后者由this序列参与组合。此外RxJS v4 保留了大量 .NET Rx 风格的别名例如select是map的别名、where是filter的别名这一点在 src/modular/index.js 中有明确实现Observable.prototype.select Observable.prototype.map; Observable.prototype.where Observable.prototype.filter;下文按类别逐一展开并为每个操作符标注调用形态与一句话语义方便按图索骥。创建 Observable 序列用途从零开始创建一条可观察序列。这类操作符几乎都是静态方法。操作符调用形态作用create静态用自定义subscribe函数创建序列是createWithDisposable的别名defer静态延迟创建每个订阅者订阅时才调用工厂函数生成序列generate静态状态驱动的循环生成序列generateWithAbsoluteTime静态按绝对时间推进的状态循环生成generateWithRelativeTime静态按相对时间推进的状态循环生成range静态生成一段连续的整数序列using静态创建与资源生命周期绑定的序列create手写订阅实现create 接受一个subscribe函数该函数收到 observer负责调用onNext/onError/onCompleted并可返回一个清理函数或 disposable 对象。完整示例var source Rx.Observable.create(function (observer) { observer.onNext(42); observer.onCompleted(); // 可选不需要清理时可以不返回 return function () { console.log(disposed); }; }); var subscription source.subscribe( function (x) { console.log(Next: x); }, function (err) { console.log(Error: err); }, function () { console.log(Completed); }); // Next: 42 // Completed subscription.dispose(); // disposeddefer 与 generate源码级实现defer 的核心逻辑在Defer.prototype.subscribeCore每次有 observer 订阅时用tryCatch调用工厂函数若抛出异常则转为observableThrow序列若返回 Promise 则自动包装为 Observable再订阅结果Defer.prototype.subscribeCore function (o) { var result tryCatch(this._f)(); if (result errorObj) { return observableThrow(result.e).subscribe(o); } isPromise(result) (result observableFromPromise(result)); return result.subscribe(o); };这解释了defer的核心价值每个订阅者都会拿到独立的、按需生成的新序列适合每次订阅都希望重新执行副作用或重新计算的场景。generate 则实现了一个状态机接收initialState初始状态、condition终止条件返回 false 结束、iterate步进函数、resultSelector结果选择器与可选的scheduler默认Scheduler.currentThread。其scheduleRecursive递归地执行判断条件 → 产出结果 → 步进状态直到条件不满足时onCompleted。典型用法var res Rx.Observable.generate( 0, // 初始状态 function (x) { return x 10; }, // 终止条件 function (x) { return x 1; }, // 步进 function (x) { return x; }); // 结果选择器range 同样支持第三个可选参数scheduler未提供时默认使用currentThreadSchedulerObservable.range function (start, count, scheduler) { isScheduler(scheduler) || (scheduler currentThreadScheduler); return new RangeObservable(start, count, scheduler); };转换事件、异步模式与数组 ⇄ Observable用途把外部世界的各类数据源DOM 事件、Node.js EventEmitter、回调、Promise、数组、可迭代对象转化为 Observable或将 Observable 收集为数组 / Map / Set / Promise。操作符调用形态作用from静态从数组、可迭代对象或类数组对象创建序列fromArray静态从数组创建序列from的底层fromCallback静态将 node 风格以外的回调函数包装为序列fromNodeCallback静态将 Node.js 风格(err, result)回调包装为序列fromEvent静态将 DOM / jQuery / EventEmitter 事件转化为序列fromEventPattern静态用自定义的 add/remove 监听函数包装事件源fromPromise静态将 Promise 转化为序列of静态将多个参数直接发射为序列toArray实例收集全部元素为数组后一次性发射toMap实例收集为Map键值对toPromise实例将序列转为 Promise取最后元素toSet实例收集为Set去重集合fromEvent一套 API 覆盖多类事件源fromEvent 是 RxJS 中最常用的事件桥接操作符签名如下Rx.Observable.fromEvent(element, eventName, [selector], [options])elementDOM 元素、NodeList、jQuery/Zepto/Angular/Ember 元素或 Node.jsEventEmittereventName事件名[selector]可选将事件参数聚合为单个对象的映射函数[options]可选事件监听选项对象。它优先使用 jQuery、Zepto、AngularJS、Ember.js 各自的绑定 API检测不到时回退到原生绑定使用 AMD 加载器时需在 RequireJS 配置中把相关库声明为 RxJS 的依赖。浏览器端示例jQueryvar input $(#input); var source Rx.Observable.fromEvent(input, click); var subscription source.subscribe( function (x) { console.log(Next: Clicked!); }, function (err) { console.log(Error: %s, err); }, function () { console.log(Completed); }); input.trigger(click); // Next: Clicked!Node.js 端示例EventEmitter selectorvar EventEmitter require(events).EventEmitter, Rx require(rx); var eventEmitter new EventEmitter(); var source Rx.Observable.fromEvent( eventEmitter, data, function (foo, bar) { return { foo: foo, bar: bar }; }); source.subscribe( function (x) { console.log(Next: foo - x.foo , bar - x.bar); }, function (err) { console.log(Error: err); }, function () { console.log(Completed); }); eventEmitter.emit(data, baz, quux); // Next: foo - baz, bar - quuxfromEvent的实现位于 src/core/linq/observable/fromevent.js其配套测试见 tests/observable/fromevent.jsfromPromise的底层实现Promise 决议后经调度器派发onNext/onError可在模块化构建的源码中追踪。组合多个 Observable 序列用途把两条或多条序列按时间或顺序关系合并为一条序列。同一功能同时提供静态与实例两种形态。操作符调用形态作用amb静态多个序列赛跑谁先发射就只响应谁prototype.amb实例同上以this序列参与赛跑combineLatest静态任一序列发射新值时用各序列最新值组合proto.combineLatest实例同上以this序列参与concat静态按顺序串联多条序列前一条完成后再订阅下一条prototype.concat实例同上以this序列开头startWith实例在序列开头预置若干元素merge静态并发交错合并多条序列prototype.merge实例同上以this序列参与mergeAll实例将序列的序列展平为单一序列repeat静态重复发射指定序列prototype.repeat实例重复发射this序列withLatestFrom实例以主序列节奏组合其他序列的最新值zip静态按索引位置一一配对多条序列的元素prototype.zip实例同上以this序列参与zip的实现src/core/linq/observable/zip.js为每条输入序列维护一个队列只有所有队列都非空时才把队首元素配对交给结果选择器否则等待——这解释了zip的按位置对齐语义而mergeAll内部对应 flatMap/selectMany 的合并器则让内层序列一完成就补位实现并发交错。函数式操作符共享副作用用途让冷序列的副作用如网络请求、定时器被多个订阅者共享避免重复执行或延迟副作用到订阅时刻。核心机制是multicastSubject 引用计数。操作符调用形态作用let实例将整个序列交给一个函数处理observable.let(fn)publish实例返回ConnectableObservableconnect()后才开始发射publishLast实例共享并只重放最后一个元素publishValue实例共享并预置一个初始值replay实例共享并把订阅前的元素缓冲重放给后订阅者share实例publish().refCount()的便捷封装订阅数归零即断开shareLast实例共享并保留最后元素给后订阅者shareReplay实例共享并以replay缓冲重放shareValue实例共享并以固定值缓存重放其中publish、replay、shareReplay等均有完整 API 文档publish、replay、sharereplay其底层RefCountDisposable与Subject的实现位于 src/core/disposables 与 src/core/subjects引用计数归零时才会真正释放底层订阅。数学与聚合操作符用途对整条序列做归约统计通常序列完成后一次性发射结果。操作符调用形态作用aggregate实例通用累加器reduce的别名形态average实例求平均值count实例统计元素个数max实例求最大值maxBy实例按 key 选择器求最大元素min实例求最小值minBy实例按 key 选择器求最小元素reduce实例从左到右累积sum实例求和这类操作符的公共比较逻辑集中在 src/core/linq/observable/_extremaby.jsmaxBy/minBy 的 key 比较与 src/core/linq/observable/_firstonly.js单元素断言序列为空时按各自语义抛出EmptyError定义见 src/core/internal/errors.js。average、sum、count、max、min均有独立实现文件与测试如 tests/observable/average.js。基于时间的操作符用途在时间维度上控制序列的发射节奏包括节流、延迟、采样、周期发射与超时控制。操作符调用形态作用debounce实例元素之后若在指定时间窗内无新元素才发射防抖debounceWithSelector实例用选择器函数决定每次的静默窗口delay实例把整条序列整体延后发射interval静态按固定周期发射递增整数sample实例按时间或另一个序列的节奏采样最新值timeInterval实例把每个元素包装为带时间间隔的对象timer静态延迟指定时间后发射可再按周期持续发射timeout实例超过指定时间无元素则报错或切换备用序列timeoutWithSelector实例用选择器动态决定每次的超时窗口timestamp实例把每个元素包装为带时间戳的对象时间操作符都依赖调度器Scheduler来安排异步任务。例如interval/timer的底层时间源实现位于 src/core/linq/observable/_observabletimer.js 与_observabletimerdateandperiod.js、_observabletimertimespanandperiod.js它们把首次延迟 周期统一换算为调度器的schedulePeriodic/scheduleFuture调用因此切换调度器如TestScheduler即可在测试中虚拟推进时间。异常处理操作符用途捕获、忽略或重试序列中的错误让错误不再一击即溃。操作符调用形态作用catch静态捕获错误并切换到备用序列prototype.catch实例同上以this序列为被保护方finally实例序列无论正常/异常结束都执行清理回调onErrorResumeNext静态忽略错误继续执行后续序列prototype.onErrorResumeNext实例同上以this序列开头retry实例出错时重新订阅可指定重试次数catch的静态实现src/core/linq/observable/catch.js内部用SingleAssignmentDisposable串行订阅各条序列遇到错误时记录lastError并继续订阅下一条全部失败后把最后一次错误抛出onErrorResumeNext则直接吞掉每个序列的错误继续推进。retry的实现会反复重订阅源序列直到成功或达到次数上限。筛选与选择操作符用途过滤、映射、截取序列中的元素这是日常使用频率最高的一类。注意表中map/filter与select/where是同一实现的两个名字见 src/modular/index.js 的别名定义。操作符调用形态作用concatMap实例映射为内层序列并按顺序串联selectConcat的别名concatMapObserver实例为 next/error/completed 分别提供映射的 concatMapelementAt实例取指定下标元素越界报错elementAtOrDefault实例取指定下标元素越界返回默认值filter实例按谓词过滤where的现代名flatMap实例映射为内层序列并展平selectMany的现代名flatMapLatest实例只保留最新内层序列selectSwitch的别名flatMapObserver实例为 next/error/completed 分别提供映射的 flatMapfind实例找第一个满足谓词的元素findIndex实例找第一个满足谓词的元素下标first实例取第一个元素可带谓词firstOrDefault实例取第一个元素空序列返回默认值includes实例判断是否包含指定元素last实例取最后一个元素可带谓词lastOrDefault实例取最后一个元素空序列返回默认值map实例对每个元素做投影变换select的现代名pluck实例提取对象元素的指定属性select实例map的传统别名selectConcat实例见concatMapselectMany实例flatMap的传统别名selectManyObserver实例flatMapObserver的传统别名selectSwitch实例见flatMapLatestsingle实例断言序列恰有一个匹配元素singleOrDefault实例同上无匹配时返回默认值skip实例跳过前 N 个元素skipLast实例跳过最后 N 个元素skipLastWithTime实例按时间跳过结尾元素skipUntil实例跳过直到另一个序列开始发射skipWhile实例跳过满足谓词的前缀take实例只取前 N 个元素takeLast实例只取最后 N 个元素takeLastBuffer实例把最后 N 个元素装进数组发射takeLastBufferWithTime实例按时间取结尾元素数组takeLastWithTime实例按时间取结尾元素takeWhile实例取满足谓词的前缀where实例filter的传统别名以flatMapselectMany为例其实现 src/core/linq/observable/selectmany.js 会对每个上游元素调用映射函数得到内层序列后立即订阅并转发其元素从而实现展平如果映射结果本身是 Promise 或可迭代对象也会被自动包装为 Observable。测试方面tests/observable/select.js 覆盖了map的参数校验、异常传播与在TestScheduler下的时间线断言可作为阅读操作符测试的范本。分组与窗口操作符用途按时间、数量或键对元素进行分组 / 打包 / 开窗输出的是序列的序列或数组的序列。操作符调用形态作用buffer实例用边界序列把元素打包成数组发射bufferWithCount实例攒满 N 个元素打包为数组bufferWithTimeOrCount实例时间或数量先到者触发打包groupBy实例按 key 选择器分组每组是一个内层序列groupByUntil实例分组并支持组的自动关闭groupJoin实例按重叠窗口做关联产生嵌套序列join实例时间窗口内的关联合并window实例用边界序列把元素切成多个子序列windowWithCount实例按数量切分子序列windowWithTime实例按时间切分子序列windowWithTimeOrCount实例时间或数量先到者切分groupBy与groupByUntil的实现位于 src/core/linq/observable/groupby.js 与 groupbyuntil.js通过一个Subject字典维护每个分组groupByUntil额外用 duration 选择器在组空闲时自动关闭并清理字典。命令式操作符用途把命令式控制流分支、循环与副作用观测融入响应式管线便于以序列的方式书写 if/while/for 逻辑。操作符调用形态作用case静态按 selector 返回值选择对应的序列switch-casedo实例副作用观测透传原值tap的传统别名doOnNext实例仅观测onNext的副作用doOnError实例仅观测onError的副作用doOnCompleted实例仅观测onCompleted的副作用doWhile实例do-while 语义先执行序列再判断条件for静态遍历数组并串联每个元素映射出的序列if静态按条件选择两条序列之一tap实例do的现代别名tapOnNext / tapOnError / tapOnCompleted实例对应doOnNext/doOnError/doOnCompleted的现代别名while静态while 语义条件满足时循环串联序列dotap的实现src/core/linq/observable/do.js在转发onNext/onError/onCompleted之前先调用观测回调且用tryCatch捕获观测回调抛出的异常并转为下游错误——这是副作用不能破坏主链路的保证。原语操作符用途发射固定语义的最小序列常用于组合测试与占位。操作符调用形态作用empty静态立即完成、不发射任何元素的空序列never静态永不发射、永不完成的序列return静态只发射一个指定值后完成throw静态只发射一个错误后终止empty与throw的实现非常直白对应 src/core/linq/observable/empty.js 与 throw.js前者在订阅时直接调度onCompleted后者直接调度onError且都接受可选scheduler参数控制派发时机。这四个原语是编写测试用例、构造占位序列时的基础积木。延伸阅读与下一步操作符全部挂载在Observable类型上建议先通读该类型文档各类操作符的完整 API 文档位于 doc/api/core/operators每个文档都包含参数说明、返回值、示例、所在源码文件src/core/linq/observable/*.js、所属发行包dist/*.js与单元测试位置tests/observable/*.js想系统学习如何用操作符构建查询链可阅读 Querying Observable Sequences若想验证操作符在时间轴上的行为参考 Schedulers 与 Testing结合TestScheduler用虚拟时间驱动interval、timer、debounce等时间类操作符仓库还提供按发行包划分的操作符清单如 rx.aggregates.md、rx.time.md可按需裁剪引入的模块。【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址: https://gitcode.com/gh_mirrors/rxj/RxJS创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考