@vueuse/rxjs 实战指南：在 Vue 3 组件中无缝集成 RxJS 响应式编程

• 前端

@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 可以看到它的依赖约定：

• peerDependencies：rxjs >= 6.0.0、vue ^3.5.0，即支持 RxJS 6/7 系列，并要求 Vue 3.5+；
• 内部依赖：仅依赖@vueuse/shared（workspace 内部包），用于tryOnScopeDispose等生命周期工具；
• 打包产物：sideEffects: false，使用 tsdown 构建，支持 ESM、IIFE（unpkg/jsdelivr）多种分发方式。

函数总览

包入口 packages/rxjs/index.ts 统一导出了 7 个 API，它们各司其职：

下面逐一深入每个 API 的用法与实现细节。

from / fromEvent：让 RxJS 工厂函数接受 ref

在纯 RxJS 中，from()无法感知 Vue 的响应式状态，fromEvent()也只能绑定已经存在的 DOM 元素。@vueuse/rxjs的 from/index.ts 对二者做了包装。

from：监听 ref 值变化

from既可以接收标准的 RxJSObservableInput，也可以接收 Vueref。传入 ref 时，它会创建一个每次 ref 值变化都会发出新值的 Observable：

import { 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 (isRef<T>(value)) return new Observable(subscriber => watch(value, val => subscriber.next(val), watchOptions))

因此第二个参数可以透传 Vue 的WatchOptions，常用选项如下：

fromEvent：等待元素就绪后自动订阅

fromEvent的增强在于：当传入的元素是一个 ref 时，即使元素在组件挂载前还是null，它也会在元素可用后自动完成订阅。源码实现里通过watch(value, ..., { immediate: true })监听元素引用，元素变为真实HTMLElement时才创建内部订阅，并妥善管理旧订阅的退订：

if (isRef<T>(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 变成 ref

useObservable/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。如果希望一开始就有确定的展示值，可以传入initialValue：

import { useObservable } from '@vueuse/rxjs' import { interval } from 'rxjs' const count = useObservable( interval(1000), { initialValue: 0 }, ) // count.value 在首次发射之前保持为 0

源码中，返回值直接以options?.initialValue初始化：

const value = deepRef<H | I | undefined>(options?.initialValue)

错误处理 onError

对于可能报错的 Observable，务必提供onError配置；否则 RxJS 会把错误视为"未处理错误"，在新调用栈中抛出并上报到window.onerror（Node 环境则是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" }, }, )

选项一览

从实现上看，useObservable的订阅对象只有next与error两个处理器，返回的是Readonly<Ref<H | I>>类型的只读 ref，值只能由流驱动、不能从组件侧直接改写。生命周期方面，它在tryOnScopeDispose中调用subscription.unsubscribe()，彻底规避了组件卸载后的内存泄漏。

useSubject：ref 与 Subject 的双向绑定

useSubject/index.ts 实现的是双向数据通道：Subject 发出的值会写入 ref，而 ref 被赋值时又会推回 Subject：

import { 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 的当前值初始化，且类型上不含undefined：

import { useSubject } from '@vueuse/rxjs' import { BehaviorSubject } from 'rxjs' const subject = new BehaviorSubject('initial') // setup() const subjectRef = useSubject(subject) // Ref<string>，而非 Ref<string | 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 变成 Observer

toObserver/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 toObserver<T>(value: Ref<T>): NextObserver<T> { 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：从组合式函数提取 Observable

useExtractedObservable/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 = shallowRef<number>() const count = useExtractedObservable( start, (start) => { return interval(1000).pipe( mapTo(1), startWith(start), scan((total, next) => next + total), takeWhile(num => num < 10), ) }, {}, { immediate: false, }, )

选项与返回值

返回值是一个只读的ShallowRef，保存提取出的 Observable 最新发射值。

从实现细节看，该函数内部用shallowRef<E | 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 与 onComplete

watchExtractedObservable(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, })

订阅选项一览

返回值：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 处理。

• 前端