2025全新RxJS入门指南：从安装到第一个响应式应用

2026-02-05 05:01:25作者：江焘钦

你还在为JavaScript异步编程中的回调地狱烦恼吗？还在为事件监听和数据处理的复杂性头疼吗？本文将带你从零开始，掌握RxJS（Reactive Extensions for JavaScript）这一强大的响应式编程库，让你轻松应对各种异步场景。读完本文，你将能够：安装并配置RxJS环境，理解核心概念Observable（可观察对象），创建并订阅简单的数据流，以及构建一个基础的响应式应用。

初识RxJS：什么是响应式编程？

RxJS是一个基于响应式编程范式的JavaScript库，它通过Observable（可观察对象）序列来组合异步操作和事件驱动的程序。简单来说，RxJS让你能够以声明式的方式处理随时间变化的数据和事件，就像处理数组一样轻松。

官方定义：A reactive programming library for JavaScript。RxJS的核心优势在于将异步操作统一为数据流，提供了丰富的操作符来转换、组合和过滤这些流，从而简化复杂异步逻辑的处理。

环境准备：安装RxJS

npm安装（推荐）

RxJS可以通过npm（Node Package Manager）轻松安装。打开终端，执行以下命令：

npm install rxjs

默认情况下，RxJS 7.x会根据不同的使用场景提供不同的代码变体：

在Node.js环境中，无论使用require还是import，都会提供针对ES5的CommonJS代码
在浏览器环境中（RxJS 7.4+），默认提供针对ES5的ES模块代码，也可以配置使用ES2015代码

如果你需要使用ES2015版本的RxJS代码，可以通过配置打包工具的es2015自定义导出条件来实现。具体配置方法请参考各打包工具文档：

其他安装方式

如果你需要使用CommonJS模块系统，或者遇到类型定义问题，可以安装补充的类型定义文件：

typings install es6-shim --ambient

对于所有模块类型（CJS/ES6/AMD/TypeScript），可以使用以下命令安装：

npm install @reactivex/rxjs

如果你使用的是npm 2版本，需要显式指定版本号：

npm install @reactivex/rxjs@7.3.0

安装指南详情：apps/rxjs.dev/content/guide/installation.md

核心概念：Observable（可观察对象）

Observable是RxJS的核心概念，它代表一个可调用的未来值或事件流的集合。Observable可以发送三种类型的通知：

next: 发送一个值（如数字、字符串、对象等）
error: 发送一个错误
complete: 通知流结束

Observable遵循"Observable语法"：next*(error|complete)?，即可以发送零到多个next通知，最后可选地发送一个error或complete通知。

创建Observable

使用Observable构造函数可以创建一个可观察对象：

import { Observable } from 'rxjs';

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

订阅Observable

要接收Observable发送的值，需要订阅它。订阅时可以提供一个包含next、error和complete方法的对象：

console.log('just before subscribe');
observable.subscribe({
  next(x) { console.log('got value ' + x); },
  error(err) { console.error('something wrong occurred: ' + err); },
  complete() { console.log('done'); }
});
console.log('just after subscribe');

执行结果：

just before subscribe
got value 1
got value 2
got value 3
just after subscribe
got value 4
done

这个例子展示了Observable可以同时处理同步和异步数据。前三个值是同步发送的，第四个值在1秒后异步发送。

Observable详细文档：apps/rxjs.dev/content/guide/observable.md

第一个响应式应用：实时数据计数器

下面我们将创建一个简单的响应式计数器应用，展示RxJS的基本用法。

功能需求

创建一个每秒递增的计数器
能够暂停和继续计数
能够重置计数器

实现代码

import { Observable, Subject, fromEvent } from 'rxjs';
import { scan, switchMap, startWith, mapTo } from 'rxjs/operators';

// 创建一个每秒发出一个值的Observable
const interval$ = new Observable(subscriber => {
  let count = 0;
  const intervalId = setInterval(() => {
    subscriber.next(count++);
  }, 1000);
  
  return () => clearInterval(intervalId);
});

// 创建控制按钮的点击事件流
const pauseButton = document.getElementById('pause');
const resumeButton = document.getElementById('resume');
const resetButton = document.getElementById('reset');

const pause$ = fromEvent(pauseButton, 'click').pipe(mapTo(false));
const resume$ = fromEvent(resumeButton, 'click').pipe(mapTo(true));
const reset$ = fromEvent(resetButton, 'click').pipe(mapTo('reset'));

// 合并控制流
const control$ = new Subject();
pause$.subscribe(() => control$.next(false));
resume$.subscribe(() => control$.next(true));
reset$.subscribe(() => control$.next('reset'));

// 构建计数器流
const counter$ = control$.pipe(
  startWith(true),
  switchMap(active => {
    if (active === 'reset') {
      return interval$.pipe(startWith(0));
    }
    return active ? interval$ : Observable.create(subscriber => {
      subscriber.next('paused');
      return () => {};
    });
  })
);

// 订阅计数器流并更新UI
const counterDisplay = document.getElementById('counter');
counter$.subscribe(value => {
  counterDisplay.textContent = value === 'paused' ? 'Paused' : value.toString();
});