RxJS 教程 - 响应式编程库

项目概述

RxJS（Reactive Extensions for JavaScript）是一个响应式编程库，用于处理异步数据流，提供了丰富的操作符来转换、组合和处理这些数据流。

项目链接：https://github.com/ReactiveX/rxjs
官方网站：https://rxjs.dev/
GitHub Stars：30k+

核心概念

Observable：可观察对象，表示一个可被订阅的数据流
Observer：观察者，用于订阅 Observable 并处理其发出的值、错误和完成信号
Subscription：订阅，表示 Observable 的执行，可用于取消订阅
Operator：操作符，用于转换、组合和处理 Observable 数据流
Subject：主题，既是 Observable 又是 Observer，可用于多播
Scheduler：调度器，控制 Observable 执行的时机和上下文

核心功能

Observable 序列：创建和处理异步数据流
操作符：丰富的操作符用于转换、组合和处理数据流
调度器：控制执行时机和上下文
错误处理：统一的错误处理机制
并发控制：处理并发操作
背压处理：处理数据流速率不匹配的情况
与多种框架集成：与 React、Angular、Vue 等框架集成
TypeScript 支持：良好的 TypeScript 类型定义
浏览器支持：支持所有现代浏览器
Node.js 支持：支持 Node.js 环境

安装与设置

基本安装

# 安装 RxJS
npm install rxjs

# 安装特定版本
npm install rxjs@7.8.1

基本设置

// 导入 RxJS
import { Observable, of, from, interval } from 'rxjs';
import { map, filter, take, tap } from 'rxjs/operators';

// 或者使用 CommonJS 导入
const { Observable, of, from, interval } = require('rxjs');
const { map, filter, take, tap } = require('rxjs/operators');

基本使用

创建 Observable

// 使用 create 创建 Observable
const observable = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  setTimeout(() => {
    subscriber.next(4);
    subscriber.complete();
  }, 1000);
});

// 使用 of 创建 Observable
const observable = of(1, 2, 3, 4);

// 使用 from 创建 Observable
const observable = from([1, 2, 3, 4]);
const observable = from(Promise.resolve('Hello'));

// 使用 interval 创建 Observable
const observable = interval(1000); // 每秒发出一个递增的数字

// 使用 fromEvent 创建 Observable
const button = document.querySelector('button');
const observable = fromEvent(button, 'click');

订阅 Observable

// 订阅 Observable
observable.subscribe(
  value => console.log('Received:', value), // 处理发出的值
  error => console.error('Error:', error), // 处理错误
  () => console.log('Completed') // 处理完成信号
);

// 简化订阅
observable.subscribe(value => console.log('Received:', value));

// 保存订阅以便后续取消
const subscription = observable.subscribe(value => console.log('Received:', value));

// 取消订阅
subscription.unsubscribe();

使用操作符

// 使用管道和操作符
const observable = of(1, 2, 3, 4, 5).pipe(
  filter(value => value % 2 === 0), // 过滤出偶数
  map(value => value * 2), // 每个值乘以 2
  take(2) // 只取前 2 个值
);

observable.subscribe(value => console.log('Received:', value));
// 输出: Received: 4, Received: 8

// 链式使用多个操作符
const observable = interval(1000).pipe(
  tap(value => console.log('Original:', value)), // 副作用操作符
  filter(value => value % 2 === 0), // 过滤出偶数
  map(value => value * 2), // 每个值乘以 2
  take(5) // 只取前 5 个值
);

observable.subscribe(
  value => console.log('Transformed:', value),
  error => console.error('Error:', error),
  () => console.log('Completed')
);

高级特性

操作符分类

RxJS 提供了丰富的操作符，主要分为以下几类：

创建操作符

import { of, from, interval, fromEvent, timer, empty, throwError } from 'rxjs';

// of: 从一组值创建 Observable
of(1, 2, 3);

// from: 从数组、Promise 或迭代器创建 Observable
from([1, 2, 3]);
from(Promise.resolve('Hello'));

// interval: 定期发出递增的数字
interval(1000);

// fromEvent: 从 DOM 事件创建 Observable
fromEvent(document, 'click');

// timer: 在指定时间后发出值

// empty: 创建一个立即完成的 Observable

// throwError: 创建一个立即抛出错误的 Observable
throwError('Error');

转换操作符

import { of } from 'rxjs';
import { map, pluck, switchMap, mergeMap, concatMap, exhaustMap } from 'rxjs/operators';

// map: 转换每个值
of(1, 2, 3).pipe(map(value => value * 2));

// pluck: 从对象中提取属性
of({ name: 'John', age: 25 }).pipe(pluck('name'));

// switchMap: 映射到新的 Observable，并取消之前的订阅
fromEvent(input, 'input').pipe(
  switchMap(value => fetch(`https://api.example.com/search?q=${value}`))
);

// mergeMap: 映射到新的 Observable，并合并所有结果

// concatMap: 映射到新的 Observable，并按顺序连接结果

// exhaustMap: 映射到新的 Observable，忽略新的请求直到当前请求完成

过滤操作符

import { of, interval } from 'rxjs';
import { filter, take, takeUntil, takeWhile, skip, debounceTime, throttleTime } from 'rxjs/operators';

// filter: 过滤值
of(1, 2, 3, 4, 5).pipe(filter(value => value % 2 === 0));

// take: 只取前 n 个值
of(1, 2, 3, 4, 5).pipe(take(3));

// takeUntil: 一直取值直到另一个 Observable 发出值
interval(1000).pipe(takeUntil(fromEvent(button, 'click')));

// takeWhile: 一直取值直到条件不满足
interval(1000).pipe(takeWhile(value => value < 5));

// skip: 跳过前 n 个值
of(1, 2, 3, 4, 5).pipe(skip(2));

// debounceTime: 防抖
fromEvent(input, 'input').pipe(debounceTime(300));

// throttleTime: 节流
fromEvent(window, 'scroll').pipe(throttleTime(1000));

组合操作符

import { of, interval } from 'rxjs';
import { combineLatest, forkJoin, merge, concat, withLatestFrom, zip } from 'rxjs/operators';

// combineLatest: 组合多个 Observable 的最新值
const observable1 = interval(1000);
const observable2 = interval(500);
combineLatest([observable1, observable2]);

// forkJoin: 等待所有 Observable 完成并收集它们的最后一个值
forkJoin([
  from(fetch('https://api.example.com/users')),
  from(fetch('https://api.example.com/posts'))
]);

// merge: 合并多个 Observable
merge(interval(1000), interval(500));

// concat: 按顺序连接多个 Observable
concat(of(1, 2, 3), of(4, 5, 6));

// withLatestFrom: 组合当前 Observable 与其他 Observable 的最新值
observable1.pipe(withLatestFrom(observable2));

// zip: 按顺序组合多个 Observable 的值
zip(observable1, observable2);

错误处理操作符

import { of, throwError } from 'rxjs';
import { catchError, retry, retryWhen, delay } from 'rxjs/operators';

// catchError: 捕获错误并返回一个新的 Observable
throwError('Error').pipe(
  catchError(error => of(`Handled error: ${error}`))
);

// retry: 出错时重试
throwError('Error').pipe(retry(3));

// retryWhen: 出错时根据另一个 Observable 决定是否重试
throwError('Error').pipe(
  retryWhen(errors => errors.pipe(delay(1000)))
);

Subject

import { Subject, interval } from 'rxjs';

// 创建 Subject
const subject = new Subject();

// 订阅 Subject
subject.subscribe(value => console.log('Subscriber 1:', value));
subject.subscribe(value => console.log('Subscriber 2:', value));

// 作为 Observer 订阅其他 Observable
const observable = interval(1000);
observable.subscribe(subject);

// 手动发出值
subject.next('Hello');
subject.next('World');

// 发出错误
subject.error('Error');

// 完成
subject.complete();

// 特殊类型的 Subject
import { BehaviorSubject, ReplaySubject, AsyncSubject } from 'rxjs';

// BehaviorSubject: 记住最后发出的值，新订阅者会立即收到该值
const behaviorSubject = new BehaviorSubject('Initial value');

// ReplaySubject: 记住指定数量的历史值，新订阅者会收到这些值
const replaySubject = new ReplaySubject(3);

// AsyncSubject: 只在完成时发出最后一个值
const asyncSubject = new AsyncSubject();

Scheduler

import { of, asyncScheduler, asapScheduler, queueScheduler } from 'rxjs';
import { observeOn, subscribeOn } from 'rxjs/operators';

// 使用 observeOn 指定观察调度器
const observable = of(1, 2, 3).pipe(
  observeOn(asyncScheduler)
);

// 使用 subscribeOn 指定订阅调度器
const observable = of(1, 2, 3).pipe(
  subscribeOn(asapScheduler)
);

// 调度器类型
// queueScheduler: 同步执行，使用队列
// asapScheduler: 尽快执行，在当前宏任务完成后
// asyncScheduler: 异步执行，使用 setTimeout
// animationFrameScheduler: 在浏览器重绘前执行

实用场景

处理用户输入

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

// 处理搜索输入
const input = document.querySelector('input');
const searchResults = document.querySelector('.results');

fromEvent(input, 'input').pipe(
  map(event => event.target.value),
  debounceTime(300), // 防抖，避免频繁请求
  distinctUntilChanged(), // 只在值变化时执行
  switchMap(query => {
    // 取消之前的请求，只处理最新的请求
    return fetch(`https://api.example.com/search?q=${query}`)
      .then(response => response.json());
  })
).subscribe(
  results => {
    // 显示搜索结果
    searchResults.innerHTML = results.map(result => `<div>${result.name}</div>`).join('');
  },
  error => {
    console.error('Error:', error);
  }
);

处理多个异步请求

import { forkJoin, from } from 'rxjs';

// 并行请求多个 API
forkJoin([
  from(fetch('https://api.example.com/users')),
  from(fetch('https://api.example.com/posts')),
  from(fetch('https://api.example.com/comments'))
]).pipe(
  map(([usersResponse, postsResponse, commentsResponse]) => [
    usersResponse.json(),
    postsResponse.json(),
    commentsResponse.json()
  ])
).subscribe(
  ([users, posts, comments]) => {
    console.log('Users:', users);
    console.log('Posts:', posts);
    console.log('Comments:', comments);
  },
  error => {
    console.error('Error:', error);
  }
);

实现拖拽功能

import { fromEvent } from 'rxjs';
import { switchMap, takeUntil, map, withLatestFrom } from 'rxjs/operators';

// 实现拖拽功能
const draggable = document.querySelector('.draggable');

const mouseDown$ = fromEvent(draggable, 'mousedown');
const mouseMove$ = fromEvent(document, 'mousemove');
const mouseUp$ = fromEvent(document, 'mouseup');

mouseDown$.pipe(
  map(event => ({
    x: event.clientX - draggable.getBoundingClientRect().left,
    y: event.clientY - draggable.getBoundingClientRect().top
  })),
  switchMap(initialOffset => {
    return mouseMove$.pipe(
      map(event => ({
        x: event.clientX - initialOffset.x,
        y: event.clientY - initialOffset.y
      })),
      takeUntil(mouseUp$)
    );
  })
).subscribe(position => {
  draggable.style.left = `${position.x}px`;
  draggable.style.top = `${position.y}px`;
});

处理 WebSocket 连接

import { webSocket } from 'rxjs/webSocket';

// 创建 WebSocket 连接
const socket$ = webSocket('wss://echo.websocket.org');

// 订阅消息
const subscription = socket$.subscribe(
  message => console.log('Received:', message),
  error => console.error('Error:', error),
  () => console.log('Connection closed')
);

// 发送消息
socket$.next('Hello WebSocket');

// 关闭连接
socket$.complete();

// 取消订阅
subscription.unsubscribe();

实现自动保存功能

import { fromEvent, interval } from 'rxjs';
import { debounceTime, distinctUntilChanged, switchMap, takeUntil, withLatestFrom } from 'rxjs/operators';

// 实现自动保存
const textarea = document.querySelector('textarea');
const saveButton = document.querySelector('button');
const status = document.querySelector('.status');

const input$ = fromEvent(textarea, 'input').pipe(
  map(event => event.target.value),
  debounceTime(1000), // 1秒无输入后保存
  distinctUntilChanged() // 内容变化时才保存
);

const save$ = input$.pipe(
  switchMap(content => {
    status.textContent = 'Saving...';
    return fetch('/api/save', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({ content })
    });
  })
);

save$.subscribe(
  response => {
    status.textContent = 'Saved!';
    setTimeout(() => {
      status.textContent = '';
    }, 2000);
  },
  error => {
    status.textContent = 'Save failed!';
  }
);

最佳实践

理解核心概念：掌握 Observable、Observer、Subscription、Operator 等核心概念
合理使用操作符：根据场景选择合适的操作符
避免内存泄漏：记得取消订阅不再需要的 Observable
错误处理：使用 catchError 等操作符处理错误
合理使用 Subject：只在需要多播时使用 Subject
性能优化：避免不必要的操作，合理使用操作符
测试：为 Observable 代码编写测试
文档：为复杂的 Observable 逻辑编写文档
与框架集成：了解如何与 React、Angular、Vue 等框架集成

常见问题与解决方案

1. 内存泄漏

问题：Observable 订阅后没有取消，导致内存泄漏

解决方案：

在组件卸载时取消订阅
使用 takeUntil 操作符自动取消订阅
使用异步管道（如 Angular 的 async pipe）

// 使用 takeUntil 自动取消订阅
import { Subject, interval } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

class Component {
  private destroy$ = new Subject();
  
  constructor() {
    interval(1000).pipe(
      takeUntil(this.destroy$)
    ).subscribe(value => console.log(value));
  }
  
  destroy() {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

2. 操作符使用不当

问题：选择了不合适的操作符，导致行为不符合预期

解决方案：

理解每个操作符的作用
查阅官方文档和示例
从小例子开始测试

3. 错误处理

问题：错误处理不当，导致整个 Observable 链中断

解决方案：

使用 catchError 操作符捕获和处理错误
合理使用 retry 操作符重试
在适当的位置处理错误

// 正确处理错误
import { throwError } from 'rxjs';
import { catchError, retry } from 'rxjs/operators';

throwError('Error').pipe(
  retry(3), // 重试 3 次
  catchError(error => {
    console.error('Error:', error);
    return of('Fallback value');
  })
).subscribe(value => console.log('Received:', value));

4. 性能问题

问题：Observable 操作导致性能问题

解决方案：

避免不必要的操作
合理使用操作符顺序
使用 share 操作符共享 Observable 执行
考虑使用更轻量级的替代方案

// 使用 share 共享 Observable 执行
import { interval } from 'rxjs';
import { share, map } from 'rxjs/operators';

const observable = interval(1000).pipe(
  map(value => value * 2),
  share() // 共享执行，避免重复计算
);

// 多个订阅者共享同一个执行
observable.subscribe(value => console.log('Subscriber 1:', value));
observable.subscribe(value => console.log('Subscriber 2:', value));

与其他异步处理方案的比较

RxJS vs Promise

单一值 vs 多个值：Promise 只能处理单一值，RxJS 可以处理多个值
不可取消 vs 可取消：Promise 不可取消，RxJS 可以通过 unsubscribe 取消
链式调用 vs 操作符：Promise 使用链式调用，RxJS 使用丰富的操作符
错误处理：Promise 使用 catch，RxJS 使用 catchError 等操作符
并发处理：RxJS 提供更多并发处理工具

RxJS vs async/await

声明式 vs 命令式：RxJS 是声明式的，async/await 是命令式的
多个值 vs 单一值：RxJS 可以处理多个值，async/await 处理单一值
操作符 vs 原生语法：RxJS 提供丰富的操作符，async/await 使用原生语法
复杂性：RxJS 学习曲线较陡，async/await 更直观

RxJS vs 回调函数

回调地狱 vs 链式操作：RxJS 避免了回调地狱，使用链式操作
错误处理：RxJS 提供统一的错误处理，回调函数需要手动处理
可读性：RxJS 代码更具可读性和可维护性
功能丰富度：RxJS 提供更多功能

参考资源

官方文档：https://rxjs.dev/guide/overview
GitHub 仓库：https://github.com/ReactiveX/rxjs
操作符文档：https://rxjs.dev/guide/operators
示例：https://rxjs.dev/examples
学习资源：https://rxjs.dev/learn
与框架集成：https://rxjs.dev/guide/integration