Angular异步核心01，再识 RxJS：Observable、Observer 与订阅的核心概念

RxJS 作为 Angular 核心依赖的响应式编程库，是理解 Angular 异步操作的关键。很多开发者在使用 Angular 时，仅停留在 “会用” HttpClient、EventEmitter 的层面，却对其底层的 Observable（可观察对象）、Observer（观察者）和订阅（Subscription）核心概念一知半解。本文将从本质出发，拆解这三个核心概念，帮你真正吃透 RxJS 的核心逻辑。

一、响应式编程：先建立基本认知

在深入概念前，先明确：RxJS 是基于响应式编程（Reactive Programming）范式的库，核心是 “数据流” 的处理 —— 把异步操作（如 HTTP 请求、DOM 事件、定时器）封装成可观察的数据流，当数据流产生变化时，观察者自动响应。

举个生活中的例子：

你（Observer）订阅了某个公众号（Observable）；
公众号发布新文章（推送数据），你就能收到通知（处理数据）；
如果你取消关注（取消订阅），就再也收不到该公众号的推送。

这个场景完美对应 RxJS 的核心三角：Observable（数据源 / 公众号）、Observer（观察者 / 你）、Subscription（订阅关系 / 关注行为）。

二、Observable：可观察的 “数据流容器”

1. 什么是 Observable？

Observable 是 RxJS 最核心的概念，翻译为 “可观察对象”，本质是一个可以异步产生多个值的数据流容器。它不会主动产生数据，只有当被订阅后，才会开始推送数据（“冷 Observable” 特性）。

2. 核心特性

惰性执行：创建 Observable 后，若没有观察者订阅，内部的逻辑不会执行；
多值推送：可推送多个值（同步 / 异步均可），区别于 Promise（仅能单次解析 / 拒绝）；
可取消：数据流可通过取消订阅终止，Promise 一旦执行无法取消。

3. 基础创建与使用

import { Observable } from 'rxjs'; // 1. 创建Observable：封装一个简单的数据流 const number$ = new Observable((subscriber) => { // subscriber 是订阅者对象，用于推送数据 subscriber.next(1); // 推送第一个值 subscriber.next(2); // 推送第二个值 // 异步推送第三个值 setTimeout(() => { subscriber.next(3); subscriber.complete(); // 标记数据流完成 }, 1000); // 取消订阅时执行的清理逻辑 return () => { console.log('Observable 已取消订阅'); }; });

三、Observer：监听数据流的 “观察者”

1. 什么是 Observer？

Observer 是一个包含回调函数的对象，用于响应 Observable 推送的不同类型通知，核心是处理 Observable 发出的数据、错误和完成信号。

2. 核心回调方法

Observer 通常包含三个可选方法：

next(value)：处理 Observable 推送的正常数据（必选，其余可选）；
error(error)：处理 Observable 抛出的错误（触发后数据流终止）；
complete()：处理 Observable 完成信号（触发后数据流终止）。

3. 基础使用示例

import { Observable } from 'rxjs'; // 定义观察者 const numberObserver = { next: (num: number) => console.log('收到数据：', num), error: (err: Error) => console.error('发生错误：', err), complete: () => console.log('数据流完成') }; // 订阅：Observable + Observer 建立关联 const subscription = number$.subscribe(numberObserver); // 输出结果（同步+异步）： // 收到数据： 1 // 收到数据： 2 // （1秒后）收到数据： 3 // 数据流完成

四、Subscription：管理订阅关系的 “凭证”

1. 什么是 Subscription？

调用Observable.subscribe()后，会返回一个 Subscription 对象，它是订阅关系的凭证，核心作用是管理订阅的生命周期（比如取消订阅）。

2. 核心方法

unsubscribe()：取消订阅，终止 Observable 与 Observer 的关联，同时触发 Observable 内部的清理逻辑；
add(subscription)：添加子订阅，便于批量取消；
remove(subscription)：移除子订阅。

3. 实战示例：取消订阅

import { Observable } from 'rxjs'; // 模拟一个持续推送的Observable const timer$ = new Observable((subscriber) => { let count = 0; const timer = setInterval(() => { subscriber.next(count++); }, 500); // 清理逻辑：取消定时器 return () => { clearInterval(timer); console.log('定时器已清除'); }; }); // 订阅并获取凭证 const timerSubscription = timer$.subscribe({ next: (num) => console.log('定时器数据：', num) }); // 3秒后取消订阅 setTimeout(() => { timerSubscription.unsubscribe(); console.log('已取消订阅'); }, 3000); // 输出结果： // 定时器数据： 0 // 定时器数据： 1 // 定时器数据： 2 // 定时器数据： 3 // 定时器数据： 4 // 定时器数据： 5 // 定时器已清除 // 已取消订阅

4. 批量管理订阅

在 Angular 组件中，常把多个订阅添加到一个 “主订阅” 中，组件销毁时批量取消：

import { Component, OnInit, OnDestroy } from '@angular/core'; import { Observable, Subscription } from 'rxjs'; @Component({ selector: 'app-demo', template: `<p>RxJS 订阅示例</p>` }) export class DemoComponent implements OnInit, OnDestroy { private mainSubscription = new Subscription(); ngOnInit(): void { // 订阅1：HTTP请求 const httpSub = this.http.get('/api/data').subscribe(); // 订阅2：定时器 const timerSub = timer$.subscribe(); // 添加到主订阅 this.mainSubscription.add(httpSub); this.mainSubscription.add(timerSub); } ngOnDestroy(): void { // 批量取消所有订阅，避免内存泄漏 this.mainSubscription.unsubscribe(); } }