前端【免费下载链接】vueuseCollection of essential Vue Composition Utilities for Vue 3项目地址https://gitcode.com/gh_mirrors/vu/vueuse点击查看免费下载vueuse/rxjs是 VueUse 生态中面向 RxJS 的官方扩展包它通过 7 个精心设计的组合式函数与工具函数把 RxJS 的Observable、Subject、Subscription与 Vue 3 的ref响应式体系桥接在一起。读完本文你将掌握在 Vue 3 组件中消费 RxJS 流、双向绑定 Subject、自动管理订阅生命周期以及用 watch 风格提取 Observable 的完整实战方案并了解每个 API 背后的源码实现原理。安装与依赖vueuse/rxjs作为 VueUse 的附加包需要与 RxJS 本体一起安装npm i vueuse/rxjs rxjs从 packages/rxjs/package.json 可以看到它的依赖约定peerDependenciesrxjs 6.0.0、vue ^3.5.0即支持 RxJS 6/7 系列并要求 Vue 3.5内部依赖仅依赖vueuse/sharedworkspace 内部包用于tryOnScopeDispose等生命周期工具打包产物sideEffects: false使用 tsdown 构建支持 ESM、IIFEunpkg/jsdelivr多种分发方式。函数总览包入口 packages/rxjs/index.ts 统一导出了 7 个 API它们各司其职函数作用from/fromEvent包装 RxJS 的from()与fromEvent()使其可以接受 VuereftoObserver把ref转换为 RxJSObserver的语法糖useObservable订阅一个Observable返回ref组件卸载时自动退订useSubject将 RxJSSubject绑定到ref双向传播值变化useSubscription使用Subscription免去手动退订、防止内存泄漏useExtractedObservable从组合式函数中提取Observable返回只读refwatchExtractedObservable以 watch 风格监听从组合式函数提取的Observable下面逐一深入每个 API 的用法与实现细节。from / fromEvent让 RxJS 工厂函数接受 ref在纯 RxJS 中from()无法感知 Vue 的响应式状态fromEvent()也只能绑定已经存在的 DOM 元素。vueuse/rxjs的 from/index.ts 对二者做了包装。from监听 ref 值变化from既可以接收标准的 RxJSObservableInput也可以接收 Vueref。传入 ref 时它会创建一个每次 ref 值变化都会发出新值的 Observableimport { from, toObserver, useSubscription } from vueuse/rxjs import { interval } from rxjs import { map, mapTo, takeUntil, withLatestFrom } from rxjs/operators import { shallowRef, useTemplateRef } from vue const count shallowRef(0) const button useTemplateRef(buttonRef) useSubscription( interval(1000) .pipe( mapTo(1), takeUntil(fromEvent(button, click)), withLatestFrom(from(count, { immediate: true, deep: false })), map(([curr, total]) curr total), ) .subscribe(toObserver(count)), // 等价于 .subscribe(val (count.value val)) )从源码看当传入的是 ref 时from内部直接构造一个Observable并通过 Vue 的watch把值变化转发给订阅者if (isRefT(value)) return new Observable(subscriber watch(value, val subscriber.next(val), watchOptions))因此第二个参数可以透传 Vue 的WatchOptions常用选项如下选项类型说明immediateboolean是否立即发出当前值deepboolean是否深度监听嵌套对象flushpre \| post \| sync回调刷新的时机fromEvent等待元素就绪后自动订阅fromEvent的增强在于当传入的元素是一个 ref 时即使元素在组件挂载前还是null它也会在元素可用后自动完成订阅。源码实现里通过watch(value, ..., { immediate: true })监听元素引用元素变为真实HTMLElement时才创建内部订阅并妥善管理旧订阅的退订if (isRefT(value)) { return new Observable((subscriber) { let innerSub: Subscription | undefined return watch(value, (element) { innerSub?.unsubscribe() if (element instanceof HTMLElement) { innerSub fromEventRx(element, event).subscribe(subscriber) subscriber.add(innerSub) } }, { immediate: true }) }) }一个典型场景按钮在组件模板中模板 ref 在setup()阶段尚未赋值但fromEvent会在挂载后自动接上事件import { fromEvent, useSubscription } from vueuse/rxjs import { useTemplateRef } from vue const button useTemplateRef(buttonRef) // 按钮元素可用时会自动订阅 useSubscription( fromEvent(button, click).subscribe(() { console.log(clicked!) }), )需要留意的是若传入的是非 ref 的null源码会直接抛出The value isnull, and it should be an HTMLElement.错误因此非 ref 场景必须传入真实元素。useObservable把 Observable 变成 refuseObservable/index.ts 是最基础的桥接函数订阅一个Observable把每次发出的值写入ref并在组件卸载时通过tryOnScopeDispose自动退订。import { useObservable } from vueuse/rxjs import { interval } from rxjs import { mapTo, scan, startWith } from rxjs/operators // setup() const count useObservable( interval(1000).pipe( mapTo(1), startWith(0), scan((total, next) next total), ), )初始值 initialValue在 Observable 发出第一个值之前ref是undefined。如果希望一开始就有确定的展示值可以传入initialValueimport { useObservable } from vueuse/rxjs import { interval } from rxjs const count useObservable( interval(1000), { initialValue: 0 }, ) // count.value 在首次发射之前保持为 0源码中返回值直接以options?.initialValue初始化const value deepRefH | I | undefined(options?.initialValue)错误处理 onError对于可能报错的 Observable务必提供onError配置否则 RxJS 会把错误视为未处理错误在新调用栈中抛出并上报到window.onerrorNode 环境则是process.on(error)import { useObservable } from vueuse/rxjs import { interval } from rxjs import { map } from rxjs/operators // setup() const count useObservable( interval(1000).pipe( map((n) { if (n 10) throw new Error(oops) return n n }), ), { onError: (err) { console.log(err.message) // oops }, }, )选项一览选项类型说明initialValueTObservable 发射前使用的初始值onError(err: any) voidObservable 错误的处理函数从实现上看useObservable的订阅对象只有next与error两个处理器返回的是ReadonlyRefH | I类型的只读 ref值只能由流驱动、不能从组件侧直接改写。生命周期方面它在tryOnScopeDispose中调用subscription.unsubscribe()彻底规避了组件卸载后的内存泄漏。useSubjectref 与 Subject 的双向绑定useSubject/index.ts 实现的是双向数据通道Subject 发出的值会写入 ref而 ref 被赋值时又会推回 Subjectimport { useSubject } from vueuse/rxjs import { Subject } from rxjs const subject new Subject() // setup() const subjectRef useSubject(subject) // 修改 subjectRef.value 会被推送到 subject subjectRef.value new value // subject 发射的值会更新 subjectRef subject.next(from subject)其双向性来源于源码中订阅与watch的组合const subscription subject.subscribe({ next(val) { value.value val }, error: options?.onError, }) watch(value, (nextValue) { subject.next(nextValue) })与 BehaviorSubject 搭配当传入的是BehaviorSubject时返回的 ref 会以 subject 的当前值初始化且类型上不含undefinedimport { useSubject } from vueuse/rxjs import { BehaviorSubject } from rxjs const subject new BehaviorSubject(initial) // setup() const subjectRef useSubject(subject) // Refstring而非 Refstring | undefined console.log(subjectRef.value) // initial源码中正是通过subject instanceof BehaviorSubject ? subject.value : undefined决定初始值并用条件类型重载让 TS 推导出精确的 ref 类型。错误处理与useObservable相同useSubject也支持可选的onError缺省时错误会被当作未处理错误上报import { useSubject } from vueuse/rxjs import { Subject } from rxjs const subject new Subject() // setup() const subjectRef useSubject(subject, { onError: (err) { console.log(err.message) // oops }, })toObserver把 ref 变成 ObservertoObserver/index.ts 是一个极简语法糖把ref包装成一个只实现next方法的NextObserver省去手写val (ref.value val)的样板代码。import { from, fromEvent, toObserver, useSubscription } from vueuse/rxjs import { interval } from rxjs import { map, mapTo, startWith, takeUntil, withLatestFrom } from rxjs/operators import { shallowRef, useTemplateRef } from vue const count shallowRef(0) const button useTemplateRef(buttonRef) useSubscription( interval(1000) .pipe( mapTo(1), takeUntil(fromEvent(button, click)), withLatestFrom(from(count).pipe(startWith(0))), map(([curr, total]) curr total), ) .subscribe(toObserver(count)), // 等价于 .subscribe(val (count.value val)) )其源码实现只有 11 行清晰体现了设计意图export function toObserverT(value: RefT): NextObserverT { return { next: (val: T) { value.value val }, } }useSubscription免管理订阅生命周期useSubscription/index.ts 接收一个实现了Unsubscribable接口的对象即 RxJS 的Subscription在组件卸载前自动调用其unsubscribe()import { useSubscription } from vueuse/rxjs import { interval } from rxjs const count ref(0) // useSubscription 会在组件卸载前调用 unsubscribe useSubscription( interval(1000) .subscribe(() { count.value console.log(count) }), )源码同样依赖tryOnScopeDispose这意味着它可以运行在组件setup()、effectScope或组合式函数内部作用域销毁时订阅自动释放无需手动管理。useExtractedObservable从组合式函数提取 ObservableuseExtractedObservable/index.ts 解决的是Observable 依赖响应式数据的问题它把组合式函数产生的ref/computed等作为源用提取函数extractor根据当前值构建 Observable并把最新发射值保存在只读 ref 中。源值变化时自动退订旧流、订阅新流组件卸载时同样自动清理。它支持与 Vuewatch全部重载签名匹配的调用形式单个源、多个源数组、as const只读数组、响应式对象均可。import { useExtractedObservable } from vueuse/rxjs import ObservableSocket from observable-socket import { computed } from vue import { makeSocket, useUser } from ../some/lib/func // setup() const user useUser() const lastMessage useExtractedObservable(user, u ObservableSocket.create(makeSocket(u.id)).down)错误与完成处理onError与useObservable语义一致额外支持onComplete在 Observable 完成时触发import { useExtractedObservable } from vueuse/rxjs import { interval } from rxjs import { mapTo, scan, startWith, takeWhile } from rxjs/operators import { shallowRef } from vue // setup() const start shallowRef(0) const count useExtractedObservable( start, (start) { return interval(1000).pipe( mapTo(1), startWith(start), scan((total, next) next total), takeWhile(num num 10), ) }, { onComplete: () { console.log(Done!) }, }, )透传 watch 选项最后一个参数可以透传 Vuewatch选项例如关闭immediate让首次源值为undefined时不执行提取// setup() const start shallowRefnumber() const count useExtractedObservable( start, (start) { return interval(1000).pipe( mapTo(1), startWith(start), scan((total, next) next total), takeWhile(num num 10), ) }, {}, { immediate: false, }, )选项与返回值选项类型说明initialValueTObservable 发射前使用的初始值onError(err: any) voidObservable 错误的处理函数onComplete() voidObservable 完成时调用返回值是一个只读的ShallowRef保存提取出的 Observable 最新发射值。从实现细节看该函数内部用shallowRefE | undefined(options?.initialValue)保存值浅层 ref避免大对象被深度代理的开销在watch回调中先退订旧订阅再调用提取器创建新流特别地源码注释指出先构造 ref 再 watch这样若 Observable 立即发射watch的后续执行能覆盖初始值。最终返回deepReadonly(obsRef)从类型层面阻止外部写入。watchExtractedObservable以 watch 风格消费提取流watchExtractedObservable/index.ts 是useExtractedObservable的命令式变体不返回 ref而是把流的最新值交给回调处理适合把流值写入组件内其他响应式状态或触发副作用。它同样支持watch的全部重载签名并在源变化与组件卸载时自动退订。典型场景音频播放器实例由模板元素动态创建其进度流需要驱动界面状态import { watchExtractedObservable } from vueuse/rxjs import { computed, reactive, useTemplateRef } from vue import { AudioPlayer } from ../my/libs/AudioPlayer // setup() const audio useTemplateRef(audio) const player computed(() (audio.value ? new AudioPlayer(audio.value) : null)) const state reactive({ progress: 0, }) watchExtractedObservable(player, p p.progress$, (percentage) { state.progress percentage * 100 })订阅选项onError 与 onCompletewatchExtractedObservable(player, p p.progress$, (percentage) { state.progress percentage * 100 }, { onError: (err: unknown) { console.error(err) }, onComplete: () { state.progress 100 // 或 0按业务需求而定 }, })透传 watch 选项第四个参数同样可以透传watch选项watchExtractedObservable(player, p p.progress$, (percentage) { state.progress percentage * 100 }, { onError: (err: unknown) { console.error(err) }, }, { immediate: true, })订阅选项一览选项类型说明onError(err: unknown) voidObservable 错误的处理函数onComplete() voidObservable 完成时调用返回值WatchHandle与 Vuewatch一致watchExtractedObservable返回一个WatchHandle可随时手动停止监听import { watchExtractedObservable } from vueuse/rxjs import { ref } from vue const source ref({ data$: null }) const stop watchExtractedObservable(source, s s.data$, (data) { console.log(data) }) // 之后需要时停止监听 stop()其实现与useExtractedObservable共享同一套watch 驱动 订阅管理骨架区别在于把next直接指向回调且不维护结果 ref因此更加轻量。综合示例点击加载文章与评论下面把多个 API 组合起来演示一个完整的实战场景——点击按钮后并行请求文章列表、为每篇文章聚合评论数与作者名并实时渲染到响应式状态中。此示例改编自包主页 README 的演示import { from, fromEvent, useObservable } from vueuse/rxjs import { forkJoin, of } from rxjs import { ajax } from rxjs/ajax import { concatAll, map, mergeMap, pluck, scan, take } from rxjs/operators import { useTemplateRef } from vue const BASE_URL https://jsonplaceholder.typicode.com const button useTemplateRef(buttonRef) const posts useObservable( fromEvent(button, click).pipe( mergeMap(() ajax.getJSON(${BASE_URL}/posts).pipe( concatAll(), take(4), mergeMap(({ id, userId, title }) forkJoin({ id: of(id), comments: ajax.getJSON(${BASE_URL}/posts/${id}/comments).pipe( map(comments comments.length), ), username: ajax.getJSON(${BASE_URL}/users/${userId}).pipe( pluck(username), ), }), 2), scan((acc, curr) [...acc, curr], []), )), ), )在这个示例中fromEvent(button, click)负责在按钮挂载后自动接上点击流useObservable把整个请求管线包装成响应式 ref流的最新结果会实时更新到模板中的posts组件卸载时订阅自动释放不会有泄漏。点击流、请求流、UI 状态三者通过vueuse/rxjs优雅地串联起来这也是 RxJS 与 Vue 组合式 API 协作的推荐范式。生命周期与内存安全的统一机制回顾全部 7 个 API会发现它们共享同一套内存安全设计凡涉及订阅的 API最终都通过vueuse/shared的tryOnScopeDispose注册退订逻辑因此无论组件卸载、effectScope结束还是组合式函数作用域销毁订阅都会被可靠清理而useExtractedObservable与watchExtractedObservable还会在 watch 回调中主动退订旧订阅确保切换数据源时旧流不会继续跑。这套机制让开发者可以专注于数据流的组装把 RxJS 的订阅管理成本完全交给 VueUse 处理。赞分享前端【免费下载链接】vueuseCollection of essential Vue Composition Utilities for Vue 3项目地址https://gitcode.com/gh_mirrors/vu/vueuse点击查看免费下载相关推荐VueUse vueuse/rxjs 的 useExtractedObservable 实战指南把 Vue 响应式状态动态映射为 RxJS ObservableVueUse vueuse/rxjs 的 useExtractedObservable 实战指南把 Vue 响应式状态动态映射为 RxJS Observab前端Airi 项目实战用 VueUse useObservable 在 Vue 组件中接入 RxJS 响应式数据流Airi 项目实战用 VueUse useObservable 在 Vue 组件中接入 RxJS 响应式数据流 导读 useObservable 是 VueUAI 应用人工智能大模型数字人AI Agent语音前端后端桌面应用移动开发即时通讯3D渲染GeoCalib训练教程使用OpenPano数据集训练自己的校准模型GeoCalib训练教程使用OpenPano数据集训练自己的校准模型 GeoCalib 是一个革命性的单图像相机校准工具它通过结合几何优化与深度学习技术能上一篇Meetily5 分钟搭好离线 AI 会议记录工具下一篇OptiScaler 指南如何免费让 AMD / Intel 显卡启用 DLSS 级超分创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考