rxjs-like-a-pro

像专家一样使用 RxJS

这项技能能帮助你编写符合惯用写法、易于组合且没有常见陷阱的 RxJS 代码。核心理念：将逻辑保留在可观察对象链中。每当你想调用 .subscribe() 时，都要想想这项工作能否改为在 .pipe() 中进行转换。

参考文件

如需了解详细示例和模式，请阅读相关参考文件：

• references/loading-state-patterns.md — 在链中派生加载／错误状态、自定义操作符 withLoadingState，以及使用 scan 在加载状态期间保留先前结果。处理需要加载指示器的异步数据获取时请阅读。
• references/massive-observable.md — 如何将臃肿的 new Observable() 构造函数重构为小而专注的部分。发现 new Observable 回调超过约 10 行时请阅读。
• references/inner-observable-chains.md — 构建包含计时、延迟和动画阶段的丰富内部可观察对象序列。组合多步骤异步序列或替换 setTimeout 模式时请阅读。
• references/custom-operators.md — 如何使用 OperatorFunction 编写内联和提取的自定义操作符。提取可复用的流逻辑时请阅读。

头号反模式：过早订阅

RxJS 中最常见的错误是过早订阅，然后在回调中执行命令式操作——在变量中跟踪状态、调用有副作用的函数，甚至更糟，在回调中再订阅另一个可观察对象（即“订阅中订阅”模式）。

这为何重要：过早订阅会让你失去响应式链的优势。你将无法再组合、重试、取消、防抖或共享这项工作。你已经脱离声明式世界，陷入命令式的杂乱代码中；每增加一项需求（添加重试、添加超时、与另一个流组合），都意味着需要更多手动管理状态。

// ❌ Bad: subscribe-in-subscribe with manual state tracking
let currentData: Data | null = null;
let loading = false;

input$.subscribe(value => {
  loading = true;
  fetchData(value).subscribe(data => {
    currentData = data;
    loading = false;
  });
});

// ✅ Good: everything is in the chain
const data$ = input$.pipe(
  switchMap(value => fetchData(value)),
);

对于加载状态，请使用 startWith 在链中派生——参见 references/loading-state-patterns.md。

臃肿的 new Observable() 反模式

另一个常见反模式是把整个程序塞进一个 new Observable(subscriber => { ... }) 构造函数中——设置监听器、解析 Promise、订阅其他可观察对象、管理重试状态，全都放在一个巨大的回调里。这是穿着可观察对象外衣的命令式代码。

new Observable() 构造函数应该小而专注——只是将一个非响应式数据源接入响应式世界的薄薄一层桥接。对于基于 Promise 的数据源，请改用 defer(() => promise)。重试逻辑、错误处理、组合数据源——这些都应该放在操作符链中。

完整的前后对比示例请参见 references/massive-observable.md。

选择合适的扁平化操作符

操作符	行为	适用场景
switchMap	新值到达时取消之前的内部可观察对象	用户输入、边输入边搜索、路由变更——只关心最新结果
mergeMap	并发运行所有内部可观察对象	需要所有结果的独立操作（日志记录、即发即弃）
concatMap	将内部可观察对象排入队列，按顺序运行	顺序很重要且不能丢弃任何内容（顺序写入、队列）
exhaustMap	内部可观察对象运行期间忽略新值	防止重复提交（表单提交点击）

对于大多数 UI／请求场景，默认使用 switchMap。

内部可观察对象不必是单个请求——它可以是使用 concat、merge、timer、delay 构成的完整事件时间线。动画和计时示例请参见 references/inner-observable-chains.md。

错误处理

如果你希望外部流继续运行，请将 catchError 放在内部可观察对象上。只有当你确实想在出错时替换整个流，才将它放在外部流上：

// ❌ Bad: catchError on outer stream kills it for good
source$.pipe(
  switchMap(value => fetchData(value)),
  catchError(err => of(fallback)),
);

// ✅ Good: catchError inside switchMap — outer stream survives
source$.pipe(
  switchMap(value =>
    fetchData(value).pipe(
      catchError(err => of(fallback)),
    ),
  ),
);

同样的原则也适用于 retry——重试内部操作，而不是整个外部流：

source$.pipe(
  switchMap(value =>
    fetchData(value).pipe(
      retry({ count: 3, delay: 1000 }),
      catchError(err => of(fallback)),
    ),
  ),
);

避免内存泄漏

手动订阅越少，发生泄漏的机会就越少。按优先级排序：

1. 完全不订阅 — 尽可能让框架处理订阅生命周期
2. 使用会自然完成的操作符 — first()、take(n)、takeUntil(destroy$)
3. 使用带通知器的 takeUntil：

const destroy$ = new Subject<void>();
someObservable$.pipe(
  takeUntil(destroy$),
).subscribe(value => /* ... */);

// In teardown: destroy$.next(); destroy$.complete();

takeUntil 必须是管道中的最后一个操作符。 它之后的操作符（尤其是扁平化操作符） 可能会创建 takeUntil 不知情的内部订阅，从而导致泄漏。

4. 组合成单个订阅 — 如果有多个带副作用的独立流， 将它们 merge 成一个流，只订阅一次。

热流与冷流

• 冷可观察对象（new Observable(...)、of()、HTTP 请求）会为每个订阅者创建一次新的执行
• 热可观察对象（Subject、fromEvent）会共享一次执行

使用 shareReplay({ bufferSize: 1, refCount: true }) 共享冷可观察对象。始终使用 refCount: true — 否则，所有订阅者取消订阅后，源订阅仍会保持活跃（内存泄漏）。

以响应式方式派生状态

不要通过多个订阅更新可变变量，而应从流中派生状态：

// ❌ Bad: mutable state, inconsistent windows
let items: Item[] = [];
let filter = '';
items$.subscribe(i => { items = i; recompute(); });
filter$.subscribe(f => { filter = f; recompute(); });

// ✅ Good: always consistent
const filteredItems$ = combineLatest([items$, filter$]).pipe(
  map(([items, filter]) => items.filter(item => item.name.includes(filter))),
);

combineLatest 与 withLatestFrom 的区别：任何输入发出时，combineLatest 都会发出（所有输入都会驱动输出）。withLatestFrom 只在源发出时才发出（一个驱动项，其他项提供上下文）。

startWith：在每个输入至少发出一次之前，combineLatest 不会发出。使用 startWith 提供初始值，让流开始工作。

谨慎使用 Subject

Subject、BehaviorSubject、ReplaySubject 是用于连接命令式代码与响应式代码的应急手段。 适用于事件总线和回调桥接。不适合作为通用状态容器 — 如果你在多个地方调用 .next() 来让 Subject 保持同步，请改用派生流。

自定义操作符

不必害怕编写自定义操作符 — 它们只是签名为 (source: Observable<A>) => Observable<B> 的函数。将重复的 .pipe() 链提取为命名操作符，并使用 OperatorFunction<In, Out>。请参阅 references/custom-operators.md，了解内联和提取的示例。

副作用应放在 tap 中，而非 subscribe 中

一个简单的经验法则：.subscribe() 不应带任何参数。所有副作用 — 日志记录、更新 DOM、 写入数据库、发送分析数据 — 都应放在链中的 tap 里。末尾的 .subscribe() 只负责激活流。

// ❌ Bad: side effects crammed into subscribe
source$.pipe(
  switchMap(value => fetchData(value)),
).subscribe(
  data => {
    updateUI(data);
    logAnalytics('data_loaded', data);
    cache.set(data);
  },
  err => showError(err),
);

// ✅ Good: side effects in tap, subscribe just activates
source$.pipe(
  switchMap(value => fetchData(value)),
  tap(data => updateUI(data)),
  tap(data => logAnalytics('data_loaded', data)),
  tap(data => cache.set(data)),
  tap({ error: err => showError(err) }),
).subscribe();

这很重要：副作用放在链中时，就可以组合。你可以添加、移除或重新排序它们。你可以在它们之间插入 filter。你可以共享流，并让不同的订阅者使用它，而不必重复副作用逻辑。如果所有内容都塞进 .subscribe()，这些优势就全都失去了。

tap 还接受带有生命周期钩子的观察者对象 — 这对调试尤其有用：

source$.pipe(
  tap({
    subscribe: () => console.log('subscribed!'),
    next: value => console.log('value:', value),
    error: err => console.log('error:', err),
    complete: () => console.log('complete'),
    unsubscribe: () => console.log('unsubscribed'),
    finalize: () => console.log('finalized (complete or unsubscribe)'),
  }),
);

subscribe 钩子对于调试“为什么我的流没有发出？”尤其有用 — 它可以确认是否确实有订阅发生。

避免不必要的 Promise 转换

firstValueFrom/lastValueFrom 适用于与基于 Promise 的 API 进行一次性互操作。在 subscribe 回调中使用它们来逃避学习响应式方法，则是代码异味 — 这类工作应放在链中，通过 switchMap 完成。

快速参考：常见重构模式

反模式	重构
a$.subscribe(x => b$.subscribe(y => ...))	a$.pipe(switchMap(x => b$))（或 mergeMap/concatMap/exhaustMap）
在 subscribe 中更新可变变量	使用 scan() 或 combineLatest 派生状态
在 subscribe 中使用 setTimeout	使用 delay()、timer() 或 debounceTime()
在 subscribe 中使用 if 守卫跳过值	在 subscribe 之前使用 filter()
在 subscribe 中使用 try/catch	在管道中使用 catchError()
手动请求取消标志	使用 switchMap（自动取消之前的请求）
多次订阅同一个冷可观察对象	shareReplay({ bufferSize: 1, refCount: true })
仅为触发副作用而调用 .subscribe()	使用 tap() 处理副作用，并让链继续执行
臃肿的 new Observable() 构造函数	使用小型且专注的构造函数 + defer() + 操作符组合
在 subscribe 中使用 await firstValueFrom()	使用 switchMap——留在链中