RxJS 响应式编程入门：从事件流到状态管理的实战指南

By | 2026年9月25日
0 Comment

为什么需要 RxJS？

作为前端开发者，我们每天都在处理异步事件：用户点击、HTTP 请求、定时器、WebSocket 消息……传统做法是使用回调或 Promise，但当多个事件源需要组合、过滤、转换时，代码会迅速变得难以维护。

RxJS 将一切视为可观察的流（Observable），提供了强大的操作符来声明式地处理异步数据。它不仅能简化复杂逻辑，还能轻松实现防抖、节流、重试、竞态取消等高级功能。

本文将通过一个实时搜索的完整示例，带你从零上手 RxJS。

环境准备

首先，创建一个新项目并安装 RxJS：


mkdir rxjs-demo && cd rxjs-demo
npm init -y
npm install rxjs

为了运行示例，我们使用 Node.js（需要支持 ES Modules）。在 package.json 中添加 "type": "module"。

核心概念：Observable、Observer 和 Subscription

Observable 代表一个可调用的未来值或事件的集合。Observer 是一个回调集合，知道如何监听 Observable 发出的值。Subscription 表示 Observable 的执行，主要用于取消执行。

让我们创建一个简单的 Observable：


import { Observable } from 'rxjs';

const observable = new Observable(subscriber => {
  subscriber.next('Hello');
  subscriber.next('RxJS');
  setTimeout(() => {
    subscriber.next('World');
    subscriber.complete();
  }, 1000);
});

const subscription = observable.subscribe({
  next: value => console.log(value),
  error: err => console.error(err),
  complete: () => console.log('完成')
});

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

输出顺序：Hello、RxJS，1 秒后 World 和 完成。

> 注意：Observable 是惰性的，只有被订阅时才会执行。每次订阅都会重新执行生产者函数。

创建操作符：从事件到流

RxJS 提供了许多创建操作符，如 of、from、fromEvent、interval 等。


import { of, from, fromEvent, interval } from 'rxjs';
import { take } from 'rxjs/operators';

// 从数组创建
from([1, 2, 3]).subscribe(console.log);

// 从事件创建（浏览器环境）
// const clicks = fromEvent(document, 'click');
// clicks.subscribe(event => console.log(event));

// 定时器，只取前 3 个值
interval(1000).pipe(take(3)).subscribe(console.log);

管道操作符：转换和过滤流

操作符是 RxJS 的灵魂。它们通过 pipe 方法组合，每个操作符接收一个 Observable 并返回一个新的 Observable。

常用操作符：

  • map：转换每个值
  • filter：过滤值
  • debounceTime：防抖
  • distinctUntilChanged：去除连续重复值
  • switchMap：映射为新的 Observable，并取消前一个

示例：


import { from } from 'rxjs';
import { map, filter } from 'rxjs/operators';

from([1, 2, 3, 4, 5])
  .pipe(
    filter(n => n % 2 === 0),
    map(n => n * 10)
  )
  .subscribe(console.log); // 20, 40

实战：实现实时搜索

假设我们有一个输入框，用户输入时触发搜索请求。我们需要：

  1. 监听输入事件
  2. 防抖，避免频繁请求
  3. 忽略空字符串
  4. 取消前一个未完成的请求，只保留最新结果

使用 RxJS 可以优雅地实现：


import { fromEvent, of } from 'rxjs';
import { debounceTime, map, distinctUntilChanged, filter, switchMap, catchError } from 'rxjs/operators';
import { ajax } from 'rxjs/ajax';

// 模拟输入元素
const input = document.createElement('input');
document.body.appendChild(input);

const search$ = fromEvent(input, 'input').pipe(
  map(event => event.target.value.trim()),
  filter(query => query.length > 0),
  debounceTime(300),
  distinctUntilChanged(),
  switchMap(query =>
    ajax.getJSON(`/api/search?q=${query}`).pipe(
      catchError(error => of({ error: true, message: error.message }))
    )
  )
);

search$.subscribe(result => {
  console.log('搜索结果:', result);
});

关键点解析：

  • debounceTime(300)：用户停止输入 300ms 后才发出值。
  • distinctUntilChanged()：如果新值与上一个值相同，则忽略。
  • switchMap：将查询字符串映射为 HTTP 请求 Observable，并自动取消前一个请求，解决竞态问题。
  • catchError：捕获错误，返回一个安全的 Observable，防止流中断。

> 💡 最佳实践：始终在 switchMap 内部处理错误，否则错误会传播并终止整个流。

常见坑与解决方案

1. 忘记取消订阅导致内存泄漏

在组件销毁时，必须取消订阅。可以使用 takeUntil 操作符：


import { Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

const destroy$ = new Subject();

someObservable.pipe(takeUntil(destroy$)).subscribe(...);

// 组件销毁时
destroy$.next();
destroy$.complete();

2. 错误处理不当

如果不在管道内捕获错误，错误会直接抛给订阅者的 error 回调，并且流会终止。对于可能出错的请求，务必使用 catchError 返回一个备用 Observable。

3. 过度使用 RxJS

不是所有异步都需要 RxJS。简单的 Promise 链可能更直观。RxJS 适合处理多个事件源的组合、取消、时间控制等复杂场景。

与 React 集成

在 React 中，我们可以使用 useEffect 订阅流，并在清理函数中取消订阅。


import React, { useState, useEffect } from 'react';
import { fromEvent } from 'rxjs';
import { debounceTime, map, distinctUntilChanged } from 'rxjs/operators';

function SearchBox() {
  const [results, setResults] = useState([]);

  useEffect(() => {
    const input = document.getElementById('search');
    const subscription = fromEvent(input, 'input')
      .pipe(
        map(e => e.target.value),
        debounceTime(300),
        distinctUntilChanged()
      )
      .subscribe(query => {
        // 模拟搜索
        fetch(`/api/search?q=${query}`)
          .then(res => res.json())
          .then(setResults);
      });

    return () => subscription.unsubscribe();
  }, []);

  return (
    <div>
      <input id="search" placeholder="搜索..." />
      <ul>
        {results.map(item => <li key={item.id}>{item.name}</li>)}
      </ul>
    </div>
  );
}

总结

RxJS 提供了一种声明式的方式处理异步流，通过操作符可以轻松实现防抖、节流、竞态取消等复杂逻辑。本文通过实时搜索的示例，展示了核心概念和最佳实践。

下一步学习方向：

  • 深入学习高阶操作符：mergeMap、concatMap、exhaustMap 的区别与使用场景。
  • 掌握 Subjects（Subject、BehaviorSubject、ReplaySubject）用于多播。
  • 探索 RxJS 在状态管理（如 NgRx、Redux-Observable）中的应用。

希望这篇教程能帮助你迈入响应式编程的大门。

> 延伸阅读： > – RxJS 官方文档 > – Learn RxJS