RxJS v4Rx.Observable.from完全指南将数组、类数组与可迭代对象转换为 Observable 序列【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址: https://gitcode.com/gh_mirrors/rxj/RxJS导读Rx.Observable.from是 RxJS v4 中把各类集合数据数组、类数组对象、字符串、ES6Set/Map以及任何符合迭代协议的对象统一转换为 Observable 序列的入口方法是日常把同步数据源接入响应式管道的首选工具。本文以仓库文档 doc/api/core/operators/from.md 为主体结合 src/core/perf/operators/from.js 与 tests/observable/from.js 的源码与测试完整讲解其参数语义、支持的数据源类型、调度器行为以及底层实现原理。读完本文你将能够熟练使用from将任意可迭代数据源拉入响应式管线并理解其枚举、映射与错误传播机制。方法签名与参数说明Rx.Observable.from(iterable, [mapFn], [thisArg], [scheduler])该方法根据一个类数组array-like或可迭代iterable对象创建一个新的 Observable 序列。各参数说明如下参数类型必填说明iterableArray/Arguments/Iterable是要转换为 Observable 序列的类数组或可迭代对象如数组、arguments、字符串、Set、Map、生成器对象等mapFnFunction否对序列中每个元素调用的映射函数签名为(value, index)thisArgAny否若提供了mapFn作为调用它时的this上下文schedulerScheduler否用于调度输入序列枚举的调度器默认Rx.Scheduler.currentThread返回值Observable—— 元素从给定可迭代序列中逐个取出的 Observable 序列。从源码看src/core/perf/operators/from.js入口函数还会做两项前置校验若iterable null抛出Error(iterable cannot be null.)若提供了mapFn但它不是函数抛出Error(mapFn when provided must be a function)。这两条校验在 src/modular/observable/from.js模块化版本中同样存在是两种构建方式共用的契约。支持的数据源类型from的灵活之处在于它能接受多种形态的输入。官方文档给出的示例覆盖了数组类对象、任意可迭代对象与字符串下面逐一说明。类数组对象arguments函数内部的arguments是典型的类数组对象可直接传入function f() { return Rx.Observable.from(arguments); } f(1, 2, 3).subscribe( function (x) { console.log(Next: x); }, function (err) { console.log(Error: err); }, function () { console.log(Completed); }); // Next: 1 // Next: 2 // Next: 3 // CompletedES6SetSet是可迭代对象按插入顺序逐个输出其元素var s new Set([foo, window]); Rx.Observable.from(s).subscribe( function (x) { console.log(Next: x); }, function (err) { console.log(Error: err); }, function () { console.log(Completed); }); // Next: foo // Next: window // CompletedES6MapMap的迭代器每次产出[key, value]二元数组因此订阅回调中收到的是键值对数组var m new Map([[1, 2], [2, 4], [4, 8]]); Rx.Observable.from(m).subscribe( function (x) { console.log(Next: x); }, function (err) { console.log(Error: err); }, function () { console.log(Completed); }); // Next: [1, 2] // Next: [2, 4] // Next: [4, 8] // Completed字符串字符串按字符逐个输出Rx.Observable.from(foo).subscribe( function (x) { console.log(Next: x); }, function (err) { console.log(Error: err); }, function () { console.log(Completed); }); // Next: f // Next: o // Next: o // Completed仅含length属性的普通对象只提供length的对象会被当作稀疏数组处理元素值为undefined此时通常配合mapFn生成实际数据。文档示例用mapFn的第二个参数索引生成数字序列// Generate a sequence of numbers Rx.Observable.from({length: 5}, function(v, k) { return k; }).subscribe( function (x) { console.log(Next: x); }, function (err) { console.log(Error: err); }, function () { console.log(Completed); }); // Next: 0 // Next: 1 // Next: 2 // Next: 3 // Next: 4 // CompletedmapFn与thisArg在枚举过程中就地映射mapFn会在元素被逐个取出的过程中被调用而不需要先产出 Observable 再叠加map操作符从而减少一次中间流转。文档示例展示了用箭头函数对元素进行加工Rx.Observable.from([1, 2, 3], function (x) { return x x; }).subscribe( function (x) { console.log(Next: x); }, function (err) { console.log(Error: err); }, function () { console.log(Completed); }); // Next: 2 // Next: 4 // Next: 6 // CompletedthisArg用于绑定mapFn的this。从实现上看映射函数并非直接使用而是经过 src/core/internal/bindcallback.js 包装当thisArg缺省时原样返回func否则按argCount 2生成function (value, index) { return func.call(thisArg, value, index); }的包装函数见 src/core/perf/operators/from.js。因此映射函数实际收到的参数是(value, index)两个。scheduler枚举运行在哪个调度器上第四个参数控制枚举过程运行的调度器。核心版perf 目录实现中的默认值是currentThreadSchedulersrc/core/perf/operators/from.js与文档中默认Rx.Scheduler.currentThread的说明一致而在模块化构建中默认值为Scheduler.queuesrc/modular/observable/from.js两者都属于在当前线程上按队列依次执行的调度策略区别在于内部调度实现。需要把枚举放到某个异步调度器上时可以显式传入例如Rx.Observable.from(array, null, null, Rx.Scheduler.timeout)这样每个元素的onNext会被安排到独立的时间片。源码级实现原理迭代器归一化getIterablesrc/core/perf/operators/from.js 中的getIterable是理解from支持多种输入的关键它按优先级分三步归一化若对象自身实现了$iterator$即 ES6 的Symbol.iterator或为旧环境 shim 的_es6shim_iterator_直接返回o[$iterator$]()的迭代器若对象是字符串包装为StringIterable按charAt逐字符产出若对象具有length属性类数组包装为ArrayIterable按索引 0..length-1 产出以上都不满足则抛出TypeError(Object is not iterable)。类数组长度的规范化toLength与maxSafeIntegerArrayIterable的长度并非直接使用o.length而是经过 src/core/perf/operators/from.js 的toLength规范化非数字或NaN视为 0、负数取 0、非有限值原样返回、超过Math.pow(2, 53) - 1maxSafeInteger的截断到该上限。这也是{ length: 5 }这类对象能被正确枚举的原因——枚举次数完全由规范化后的length决定。递归调度与错误传播createScheduleMethodFromObservable.subscribeCore获取迭代器后调用scheduler.scheduleRecursive(0, ...)开始递归调度src/core/perf/operators/from.js。每一轮递归createScheduleMethod生成的loopRecursive用tryCatch(it.next).call(it)取下一个元素若迭代器本身抛错直接onError若next.done为真调用onCompleted()正常结束若有映射函数用tryCatch(fn)(result, i)做映射映射过程抛错同样转为onError否则onNext(result)并递归到i 1。这种迭代一步、调度一步的设计意味着元素是惰性逐个产出的mapFn抛出的异常例如throw new Error(woops)会被捕获并转换为序列的onError通知而不会中断整个订阅。用测试验证行为仓库中的单元测试 tests/observable/from.js 使用 QUnit TestScheduler精确验证了上述全部行为可作为行为契约参考from Array[1,2,3,4,5]依次在时间 201~205 发出1..5206 完成from Array Empty空数组立即完成onCompleted(201)from With Length{ length: 5 }配合(v, k) k产出 0..4from With Stringfoo依次产出f、o、ofrom With Selector / Selector Error / Selector Some Error验证映射函数正常映射、整体抛错直接onError、中途抛错时先发出已映射元素再onErrorfrom With Selector And Context验证thisArg被正确绑定equal(this, context)from With Map / from With Set在环境支持迭代协议含 Mozillaiterator兼容分支时验证Map输出键值对数组、Set输出元素。与fromArray、of的关系在核心版实现中还保留了Rx.Observable.fromArraysrc/core/perf/operators/fromarray.js它只接受数组与调度器两个参数其注释明确标注deprecated use Observable.from or Observable.of。也就是说from是更通用、更推荐的替代方案fromArray能做的数组 → Observablefrom都能做且还额外支持字符串、Set、Map、生成器与任意类数组对象。在仓库中的获取方式本文所述实现存在两处对应源码核心版位于 src/core/perf/operators/from.js模块化CommonJS版本位于 src/modular/observable/from.js。对应的预构建产物位于 modules 目录例如rx-lite、rx-core等模块包中均已包含from实现其中模块化版本的测试位于 src/modular/test/from.js。文档本身doc/api/core/operators/from.md还列出了from被包含在rx.all、rx、rx.lite等完整构建中并可通过 NPM 包rx与 NuGet 的RxJS-Complete、RxJS-Main、RxJS-Lite获取。小结Rx.Observable.from(iterable, [mapFn], [thisArg], [scheduler])是把同步集合数据接入响应式世界的统一入口它通过getIterable归一化数组、类数组、字符串与任意可迭代对象通过scheduleRecursive实现惰性逐个枚举通过tryCatch保证迭代与映射过程中的异常都能转为onError通知并通过mapFn/thisArg支持在枚举同时完成映射。掌握它即可将任意数据集合平滑地变为可订阅、可组合、可调度的 Observable 序列。【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址: https://gitcode.com/gh_mirrors/rxj/RxJS创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
