RxJS 4 `partition` 操作符深度解析：按谓词将一条 Observable 流一分为二

后端【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址https://gitcode.com/gh_mirrors/rxj/RxJS点击查看免费下载本文基于 RxJS v4The Reactive Extensions for JavaScript官方 API 文档与仓库源码系统讲解Rx.Observable.prototype.partition(predicate, [thisArg])的语义、参数、底层实现与测试验证。读完本文你将掌握如何用一条谓词函数把任意 Observable 拆分成两个互斥的输出流命中/未命中并理解其惰性订阅、错误传播与完成语义在源码层面的由来。一、partition 是什么API 定位与核心语义partition是 RxJS 4 中一个极其实用的分流操作符。它接收一个谓词predicate函数将源 Observable 产生的所有观测值按该谓词的判定结果拆分成两个 Observable第一个 Observable触发所有谓词返回true的观测值第二个 Observable触发所有谓词返回false的观测值。官方文档对它的定义如下原文见 doc/api/core/operators/partition.mdReturns two observables which partition the observations of the source by the given function. The first will trigger observations for those values for which the predicate returns true. The second will trigger observations for those values where the predicate returns false.这句话包含了四个关键语义理解它们是正确使用partition的前提返回的是数组而非单个 Observable——返回类型为Array下标 0 是命中分支下标 1 是未命中分支判定是互斥且完备的——每个值恰好落入两个分支之一不存在漏掉或重复的情况等价于filter(predicate)与filter(!predicate)的组合谓词针对每个订阅者独立执行一次——原文明确指出 The predicate is executed once for each subscribed observer。这意味着如果你订阅了第一个输出又订阅了第二个输出谓词会分别对两次订阅各自评估而不是共享一次判定结果错误与完成原样传播——Both also propagate all error observations arising from the source and each completes when the source completes。源流抛出的错误会同时送达两个分支的订阅者源流完成时两个分支也各自完成。方法签名Rx.Observable.prototype.partition(predicate, [thisArg])二、参数详解1.predicateFunction必填选择器函数对源流产生的每个元素调用一次用于决定该元素进入哪个输出分支。回调调用时携带以下三个参数参数含义value当前元素的值index当前元素在源序列中的索引从 0 开始observable正在被订阅的 Observable 对象本身2.thisArgAny可选执行predicate时用作this的对象。不传时谓词按普通函数调用。返回值Array类型包含两个 Observable数组第 0 项谓词返回true时触发数组第 1 项谓词返回false时触发。三、源码级实现剖析partition 到底做了什么partition的实现极其简洁却包含了整个操作符的全部语义。核心源码位于 src/core/linq/observable/partition.jsobservableProto.partition function(predicate, thisArg) { var fn bindCallback(predicate, thisArg, 3); return [ this.filter(predicate, thisArg), this.filter(function (x, i, o) { return !fn(x, i, o); }) ]; };从源码可以看出三个事实事实一partition 是两次 filter 的组合。第一个输出分支直接复用filter(predicate, thisArg)第二个分支则用一个包装函数对谓词结果取反!fn(x, i, o)。这与返回的数组第一个为真、第二个为假的文档语义完全吻合也意味着你可以把partition(p)等价理解为[source.filter(p), source.filter(x !p(x))]的语法糖。事实二thisArg 通过 bindCallback 统一绑定。源码第一行调用bindCallback(predicate, thisArg, 3)生成绑定后的判定函数。其实现位于 src/core/internal/bindcallback.jsvar bindCallback Rx.internals.bindCallback function (func, thisArg, argCount) { if (typeof thisArg undefined) { return func; } switch(argCount) { case 3: return function(value, index, collection) { return func.call(thisArg, value, index, collection); }; // ... 其他 argCount 分支 } return function() { return func.apply(thisArg, arguments); }; };注意argCount传入的是3正好对应文档声明的谓词三参数(value, index, observable)。当不传thisArg时bindCallback直接返回原始函数零额外开销。事实三惰性求值——partition 本身不执行任何判定。partition调用时只是构造出两个 filter 链没有任何订阅发生。真正执行谓词发生在后续对输出分支调用subscribe时。这也解释了谓词对每个订阅者执行一次两次独立的订阅会各自走一遍 filter 的判定流程。深入 filter判定异常如何被处理第二个分支对谓词取反后最终都会落到 filter 的实现上。模块化版本见 src/modular/observable/filter.js其中FilterObserver是判定真正发生的场所FilterObserver.prototype.next function(x) { var shouldYield tryCatch(this._fn)(x, this._i, this.source); if (shouldYield errorObj) { return this._o.onError(shouldYield.e); } shouldYield this._o.onNext(x); };这里有两个值得注意的细节索引由 filter 内部维护this._i并连同this.source一并传给谓词——这就是文档中index与observable两个回调参数的来源谓词抛出的异常会被 tryCatch 捕获并转为 onError送达下游而不会中断整个订阅链。这正对应了文档propagate all error observations arising from the source中错误的另一个来源不仅源流自身的onError会传播谓词求值阶段的异常同样会走错误通道。模块化版本仓库中的模块化实现位于 src/modular/observable/partition.js与核心版逻辑一致只是以 CommonJS 模块形式导出module.exports function partition (source, predicate, thisArg) { var fn bindCallback(predicate, thisArg, 3); return [ filter(source, predicate, thisArg), filter(source, function (x, i, o) { return !fn(x, i, o); }) ]; };四、完整示例4.1 ES6 解构奇偶分流这是官方文档给出的第一个示例ES6 语法利用数组解构直接拿到两个分支let [evens, odds] Rx.Observable.range(0, 10) .partition(x x % 2 0); let subscription1 evens.subscribe( x console.log(Evens: %s, x), e console.log(Error: %s, e), () console.log(Completed) ); // Evens: 0 // Evens: 2 // Evens: 4 // Evens: 6 // Evens: 8 // Completed let subscription2 odds.subscribe( x console.log(Odds: %s, x), e console.log(Error: %s, e), () console.log(Completed) ); // Odds: 1 // Odds: 3 // Odds: 5 // Odds: 7 // Odds: 9 // CompletedRx.Observable.range(0, 10)产生 09 十个整数x % 2 0把偶数分给evens、奇数分给odds且两个分支最终都会收到Completed通知——这正是each completes when the source completes的直观体现。4.2 DOM 事件鼠标位置左右分区官方文档的第二个示例展示了一个非常典型的 UI 应用场景——把鼠标移动事件按屏幕中线拆成左右两个数据流完整 HTML 骨架与 JS 如下div iddom-event-source/div div iddom-event-output div classleftp/p/div div classrightp/p/div /divvar sourceElement document.getElementById(dom-event-source); var elements [#dom-event-output .left p, #dom-event-output .right p].map(document.querySelector.bind(document)); var elementRect sourceElement.getBoundingClientRect(); var observers Rx.Observable.fromEvent(sourceElement, mousemove) .map(e ({ x: Math.floor(e.clientX - elementRect.left), y: Math.floor(e.clientY - elementRect.top) }) ).partition(pos pos.x sourceElement.clientWidth / 2); elements.forEach((n, i, a) observers[i].subscribe(displayCoordinates.bind(displayCoordinates, n))); function displayCoordinates(element, pos) { element.textContent (x: ${pos.x}, y: ${pos.y}); }这个例子的巧妙之处在于先用map把原始的mousemove事件归一化成{ x, y }坐标对象再用partition(pos pos.x sourceElement.clientWidth / 2)按横坐标是否小于元素宽度一半来分流左半区进observers[0]右半区进observers[1]两个输出分支分别订阅各自把坐标渲染到对应的p标签中实现鼠标在左显示左边、在右显示右边的联动效果。这里也示范了partition一个重要的工程用法用一条判定函数同时驱动两个独立的 UI 分支避免在回调里手写 if/else 分发逻辑。五、测试验证行为边界与订阅语义仓库在 tests/observable/partition.js 中为partition提供了 7 组基于TestScheduler的 QUnit 测试覆盖了空序列、单元素、多元素、完成、未完成、错误、中途取消订阅等全部关键场景测试名场景断言要点partition empty空序列210 完成两分支均在 210 收到onCompleted且各自对源发起了订阅subscribe(200, 210)× 2partition single仅一个元素 4分支 1 收到onNext(210, 4)后完成分支 2 直接完成未收到任何值partition each元素 4、3分支 1 收到 4分支 2 收到 3两者同时在 230 完成partition completed多个元素后 360 完成两分支按奇偶正确分流并同时收到完成通知partition not completed源永不完成两分支各自只收到命中的值订阅持续到 1000虚拟时间终点partition error290 时刻源抛错两分支同时在 290 收到同一个error随后双双终止subscribe(200, 290)× 2partition disposed280 时刻取消两个订阅两分支停止接收后续值源的两个订阅均在 280 被释放这些测试精确印证了文档描述的核心语义错误分支onError不是只发给命中或未命中的某一方而是广播给两个分支——这对应 tests/observable/partition.js 中partition error的断言独立订阅每次对输出分支调用subscribe都会对源产生一次新的订阅xs.subscriptions断言始终包含两条subscribe记录这就是谓词对每个订阅者执行一次的底层原因完成/终止的同步性无论源完成、报错还是被取消订阅两个分支的终止时刻始终保持一致。六、获取方式与适用环境源码位置核心实现src/core/linq/observable/partition.js模块化实现src/modular/observable/partition.js底层依赖src/core/internal/bindcallback.js、src/modular/observable/filter.js单元测试tests/observable/partition.js构建产物按官方文档说明partition包含在以下构建产物与发行渠道中对应仓库内 modules 目录下的模块化构建以及 npmrx包、NuGet 的RxJS-All、RxJS-Coincidence包完整版rx.all与兼容版rx.all.compatrx.coincidence模块版本注意事项本文基于RxJS v4即 Reactive Extensions for JavaScript 的 4.x 系列文档头部明确标注 This is RxJS v 4。如果你使用的是新一代 RxJS5 的rxjs包其 API 形态如partition操作符的位置与导入方式已有较大差异请以对应版本的官方文档为准。七、小结什么时候该用 partitionpartition适合一切把一条流按布尔条件拆成两条并行流的场景UI 分流如按事件坐标、状态值把渲染逻辑拆成两路见 4.2 的鼠标分区示例数据归类如按奇偶、按阈值、按类型把数据流拆成两路分别消费见 4.1替代 if/else 分发相比在订阅回调里手写条件分支partition让两个分支各自拥有独立的订阅链可以分别施加不同的后续操作符组合与错误处理代码结构更清晰。记住它的三条核心语义即可放心使用两分支互斥且完备、谓词按订阅者独立求值、错误与完成同步传播到两路。赞分享后端【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址https://gitcode.com/gh_mirrors/rxj/RxJS点击查看免费下载相关推荐es-toolkit 迭代器 partition 详解单遍扫描将元素按谓词拆分为两组es toolkit 迭代器 partition 详解单遍扫描将元素按谓词拆分为两组 partition 是 es toolkit 迭代器模块中的一个终端操作前端后端rxjs/observable-polyfill 深度解析RxJS 9 的 Web 平台 Observable 条件回退实现rxjs/observable polyfill 深度解析RxJS 9 的 Web 平台 Observable 条件回退实现 导读 rxjs/observ前端RxJS 4 amb 操作符全解析让最先响应的 Observable 或 Promise 胜出RxJS 4 amb 操作符全解析让最先响应的 Observable 或 Promise 胜出 amb Ambiguous歧义消除是 RxJS 4 中用后端上一篇终极 Markdown 解析工具 mdq从安装到精通的完整路线图下一篇Watermill安全审计消息流中的合规性与数据保护创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考