RxJS

RxJS 简介

RxJS（Reactive Extensions for JavaScript）是一个 响应式编程库，用于处理 异步数据流和事件驱动编程。它基于 观察者模式（Observer Pattern） 和 函数式编程（Functional Programming），提供了一套强大的工具来管理 事件、异步请求、状态变更 等复杂数据流。RxJS 的核心是 Observable（可观察对象），允许开发者以声明式的方式组合和转换数据流，适用于 前端应用、Node.js 后端和实时数据处理。

核心特性

1. Observable 和 Observer
   • Observable：表示异步数据流（如 HTTP 请求、用户输入）
   • Observer：订阅 Observable 并处理数据（next、error、complete）
2. 操作符（Operators）
   • 转换：map、filter、reduce
   • 组合：merge、concat、switchMap
   • 错误处理：catchError、retry
3. Subject 和 Multicasting
   • Subject：既是 Observable 又是 Observer，支持多播（如 BehaviorSubject、ReplaySubject）
4. 调度器（Schedulers）
   • 控制异步任务的执行时机（如 asyncScheduler、queueScheduler）
5. 取消订阅（Unsubscribe）
   • 通过 Subscription 管理资源释放，避免内存泄漏
6. 与框架集成
   • 支持 Angular（默认内置）、React、Vue
7. 高性能与惰性计算
   • 操作符链式调用按需执行

核心概念

典型应用场景

1. 用户输入处理
   • 防抖搜索（debounceTime）、连续点击限制
2. HTTP 请求管理
   • 取消重复请求（switchMap）、错误重试
3. 状态管理
   • 结合 Redux 或独立使用（如 BehaviorSubject 存储状态）
4. 实时数据流
   • WebSocket 消息、传感器数据聚合
5. 动画与游戏
   • 帧率控制（animationFrameScheduler）

快速入门示例

1. 安装

npm install rxjs

2. 基础 Observable

import { Observable } from 'rxjs';

const observable = new Observable(subscriber => {
  subscriber.next(1);
  setTimeout(() => subscriber.complete(), 1000);
});

observable.subscribe({
  next: value => console.log(value), // 1
  complete: () => console.log('Done')
});

3. 操作符链式调用

import { fromEvent } from 'rxjs';
import { debounceTime, map } from 'rxjs/operators';

fromEvent(document, 'input')
  .pipe(
    debounceTime(300),
    map(e => e.target.value)
  )
  .subscribe(value => console.log(value));

4. Subject 多播

import { BehaviorSubject } from 'rxjs';

const subject = new BehaviorSubject('初始值');
subject.subscribe(value => console.log(value)); // 立即输出"初始值"
subject.next('新值'); // 输出"新值"

性能优化建议

1. 取消无用订阅
   • 使用 takeUntil 或手动 unsubscribe()
2. 避免嵌套 Observable
   • 用 switchMap/mergeMap 替代嵌套 subscribe
3. 共享数据流
   • 多播（shareReplay）避免重复计算
4. 按需引入操作符
   • 使用 rxjs/operators 减少打包体积
5. 调度器优化
   • 高频率事件用 animationFrameScheduler