企业官网建设流程全解析

RxJS Subscription 完全指南：订阅生命周期、取消订阅与资源释放机制

【免费下载链接】rxjsA reactive programming library for JavaScript项目地址: https://gitcode.com/gh_mirrors/rx/rxjs

本指南围绕 rxjs.dev 官方文档 guide/subscription.md 展开，结合当前仓库packages/rxjs的源码实现，深入讲解 Subscription（订阅）的概念、unsubscribe()的取消机制、add/remove的组合订阅管理，以及现代 RxJS 内核中基于 Subscriber 与 teardown 回调的资源释放模型。读完本文，你将掌握如何正确创建、组合、取消订阅，并理解取消订阅在底层是如何真正“停止”一个 Observable 执行的。

什么是 Subscription？

Subscription 是一个表示可释放资源（disposable resource）的对象，通常对应一次 Observable 的执行。它最重要的方法只有一个——unsubscribe()，该方法不接收任何参数，调用后即释放该订阅所持有的资源。

在早期版本的 RxJS 中，Subscription 被命名为 "Disposable"（可丢弃对象），这个命名更直白地传达了它的本质：订阅是一次性的、可终结的资源句柄。

import { interval } from 'rxjs'; const observable = interval(1000); const subscription = observable.subscribe(x => console.log(x)); // Later: // This cancels the ongoing Observable execution which // was started by calling subscribe with an Observer. subscription.unsubscribe();

如上例所示，interval(1000)每秒发出一个递增数字。调用subscribe()会启动一次独立的 Observable 执行，unsubscribe()则终止这次执行——此后不再有任何值被推送，相关的定时器与内部状态也被清理。

一个 Subscription 本质上只做一件事：提供一个unsubscribe()函数，用于释放资源或取消 Observable 的执行。

unsubscribe()：一次订阅，一个执行

理解unsubscribe()的关键在于"每次订阅都是独立执行"。同一个 Observable 可以被多次订阅，每次订阅都会创建独立的生产者执行和独立的 Subscriber。

这在当前仓库的 ColdObservable 实现 中有清晰的印证：

subscribe(maybeObserver?: Partial<Observer<T>> | ((value: T) => void) | null, config?: SubscribeOptions): void { const subscriber = new ColdSubscriber(maybeObserver, config?.signal); try { this.#init(subscriber); } catch (error) { subscriber.error(error); } }

源码注释也明确写道：ColdObservable 的每次直接subscribe()调用都会创建一个"独立的生产者执行和 Subscriber"（independent producer execution and Subscriber）。因此：

• 取消订阅 A 不影响 B；
• 取消订阅只能作用于它自己那一次执行；
• 想让多次订阅全部停止，要么逐个调用unsubscribe()，要么使用下文介绍的组合订阅。

组合订阅：add 与 remove

Subscription 还支持组合：可以把一个订阅“添加”到另一个订阅上，这样对父订阅调用一次unsubscribe()，就能级联取消所有子订阅。官方文档给出了经典的定时器示例：

import { interval } from 'rxjs'; const observable1 = interval(400); const observable2 = interval(300); const subscription = observable1.subscribe(x => console.log('first: ' + x)); const childSubscription = observable2.subscribe(x => console.log('second: ' + x)); subscription.add(childSubscription); setTimeout(() => { // Unsubscribes BOTH subscription and childSubscription subscription.unsubscribe(); }, 1000);

运行结果如下：

second: 0 first: 0 second: 1 first: 1 second: 2

interval(300)节奏更快，所以 "second" 先出现；setTimeout(..., 1000)触发后，一次subscription.unsubscribe()同时终止了observable1和observable2两条执行链（注意输出里没有 "first: 2"，因为 400ms 周期的第二次发出发生在 800ms，第三次在 1200ms——已在 1000ms 被取消）。

与add对应的是remove(otherSubscription)方法，用于撤销之前添加的子订阅：

subscription.remove(childSubscription); // 之后调用 subscription.unsubscribe() 只取消父订阅，不再影响 childSubscription

组合订阅的典型应用场景是：一个页面/组件持有多个订阅，统一挂到一个"根订阅"上，销毁时一次清理，避免逐个unsubscribe()的遗漏。

源码级原理：Subscriber、teardown 与信号模型

当前仓库中的packages/rxjs采用新一代内核架构（Symbol 操作符 + 平台 Observable 互操作），经典Subscription类的能力被内化到Subscriber与 teardown 回调模型之中。理解这一层，能帮你真正读懂"取消订阅时发生了什么"。

ColdSubscriber：每次订阅的资源容器

ColdSubscriber 是冷 Observable 每次订阅时创建的订阅者对象，它内部维护了三样关键状态：

class ColdSubscriber<T> implements Subscriber<T> { #abortController = new AbortController(); #destination: Partial<Observer<T>> | null = null; #teardowns: (() => void)[] | null = null; readonly #signal: AbortSignal; get active() { return !this.#signal.aborted; } get signal() { return this.#signal; } // ... }

• #abortController：内部取消控制器，error/complete时会abort()；
• #teardowns：注册的清理回调（teardown）队列，abort 时依次执行；
• #signal：由父级信号与本控制器信号通过AbortSignal.any合成，active即"未中止"。

构造函数中注册了 abort 监听器：一旦信号中止，就取出并清空 teardown 队列，逐个执行清理逻辑。这正是unsubscribe()的现代等价物——取消订阅 → 信号 abort → 依次执行所有 teardown 回调。

addTeardown：注册资源清理钩子

操作符通过subscriber.addTeardown(teardown)注册自己的资源清理逻辑。仓库中有大量实证：

• animation-frames.ts：注册 teardown，在取消订阅时调用cancelAnimationFrame(id)停止下一帧请求；
• async-subject.ts：注册 teardown，把订阅者从内部订阅集合中移除；
• buffer-time.ts：注册 teardown 清理定时器与缓冲上下文；
• tap.ts：注册 teardown，在订阅被显式取消时触发unsubscribe与finalize钩子。

以tap为例，它的生命周期钩子定义中明确包含三个回调（tap.ts）：

export interface TapObserver<T> extends Observer<T> { subscribe: () => void; unsubscribe: () => void; finalize: () => void; }

其测试用例 tap.spec.ts 验证了完整生命周期：显式unsubscribe时依次触发unsubscribe与finalize；自然完成（complete）时只触发finalize，不触发显式取消钩子。这印证了"取消订阅"与"自然结束"在资源清理上走的是不同路径。

signal：现代取消机制

新内核中，取消订阅还可以通过AbortSignal完成。工具函数 subscribeToSource 展示了信号如何向上游传播：

source.subscribe( { next: guard<[T]>(overrides?.next, (value) => subscriber.next(value)), error: handleError, complete: guard<[]>(overrides?.complete, () => subscriber.complete()), }, { signal: signal ? AbortSignal.any([subscriber.signal, signal]) : subscriber.signal } );

下游订阅者信号通过AbortSignal.any与操作符局部信号合并后传给上游，从而形成一条可级联传播的取消链。调用方也能主动使用AbortController取消订阅，例如 tap 测试中的写法：

const firstController = new AbortController(); tapped.subscribe((value) => firstResults.push(value), { signal: firstController.signal }); // ... firstController.abort();

实战模式：用 takeUntil 声明式结束订阅

除了手动持有 Subscription 并调用unsubscribe()，更符合 RxJS 风格的是声明式取消——通过操作符让订阅在条件满足时自动终止。

当前仓库中takeUntil的实现（take-until.ts）非常清晰地演示了这一模式：

Observable.prototype[takeUntil] = function <T>(this: Observable<T>, notifier: ObservableValue<any>): Observable<T> { return thiscreate => { let notifierSource: Observable<any>; try { notifierSource = Observable.from(notifier); } catch (error) { subscriber.error(error); return; } subscribeToSource(notifierSource, subscriber, { next: () => subscriber.complete(), complete: () => void 0 }); if (!subscriber.active) { return; } subscribeToSource(this, subscriber); }); };

其原理是：同时订阅源与 notifier；notifier 一旦发出值，就调用subscriber.complete()，从而触发整个链条的 teardown 清理——订阅自然终结，无需显式unsubscribe()。

组件销毁时常见的写法：

// 借助 notifier 统一结束所有订阅 const destroy$ = new Subject<void>(); source.pipe(takeUntil(destroy$)).subscribe(/* ... */); // 组件销毁时 destroy$.next(); destroy$.complete();

这与手动subscription.unsubscribe()是两种互补的策略：命令式取消适合精确控制单个订阅，**声明式取消（takeUntil 等）**适合按事件/生命周期统一收尾，且不会出现遗漏。

最佳实践小结

1. 有订阅必有取消：任何长期存在的 Observable（interval、timer、事件监听等）订阅后都要在合适的时机取消，否则会泄漏资源。仓库的 teardown 机制（如cancelAnimationFrame、移除订阅者集合）都依赖取消动作触发。
2. 组合订阅统一管理：把多个订阅add到一个根 Subscription 上，销毁时一次unsubscribe()清理全部；remove可在运行期动态解除子订阅。
3. 优先声明式取消：在组件/页面生命周期明确的场景，takeUntil(destroy$)比散落的手动unsubscribe()更不易出错。
4. 理解自然终结与显式取消的差异：从源码看，complete()/error()会 abort 并执行 teardown，但 tap 的unsubscribe钩子仅在显式取消时触发（见 tap.ts 与 tap.spec.ts），据此安排资源回收逻辑。
5. 善用 AbortSignal：新内核中订阅接受{ signal }配置项，AbortController.abort()是unsubscribe()之外另一种标准化的取消方式（见 observable-helpers.ts）。

延伸阅读

• 官方指南：guide/observable.md（Observable 执行模型）、guide/subject.md（Subject 与多播）
• 内核实现：cold-observable.ts（ColdSubscriber/teardown 核心）、create.ts（派生 Observable 构造协议）
• 资源释放实证：animation-frames.ts、tap.ts、take-until.ts 及其对应*.spec.ts测试

【免费下载链接】rxjsA reactive programming library for JavaScript项目地址: https://gitcode.com/gh_mirrors/rx/rxjs