web-utilities-rxjs

RxJS 响应式编程模式

快速指南： RxJS 将异步工作组合成流。Observable 是惰性的——直到调用 subscribe() 才会运行——而 pipe() 中的操作符描述的是管道，而不是执行管道。大多数情况下，关键在于做出三个决定：选择哪个扁平化操作符（switchMap 会取消，mergeMap 会并行处理，concatMap 会排队，exhaustMap 会忽略）、catchError 放在哪里（放在内部 管道中；否则错误会永久终止外部流），以及如何结束订阅。Observable 变量名以 $ 结尾。

详细资源：

• examples/core.md — 创建、使用 defer 延迟执行、转换、错误恢复、Promise 互操作
• examples/higher-order.md — 通过实例讲解 switchMap、mergeMap、concatMap、exhaustMap
• examples/subjects.md — 四种 Subject 类型、多播、shareReplay
• examples/combination.md — combineLatest、forkJoin、merge、concat、race
• examples/memory-leaks.md — takeUntil 模式、订阅集合、自动完成的流
• reference.md — 操作符表、场景到操作符的查找表、弃用项

---

适用路径

清理方式取决于源是否会自行结束。

• 源会自行完成 — of、from、包装后的 Promise、timer(n)、单次 HTTP 响应。无需取消订阅；直接参阅 examples/core.md。
• 源会持续运行，直到被停止 — fromEvent、interval、任何 Subject、WebSocket。每个订阅都需要结束方式，相关模式见 examples/memory-leaks.md。
• 需要从流中获取一个值 — firstValueFrom / lastValueFrom 可桥接到 await，并代为结束生命周期。参阅 examples/core.md。

---

<critical_requirements>

编写 RxJS 代码前

为每个长期订阅设置结束条件——使用带销毁 Subject 的 takeUntil、Subscription 集合，或 take、first 等会自行完成的操作符。没有结束条件的 fromEvent 或 interval 订阅会比创建它的对象存活得更久，导致其闭包无法释放。

将 takeUntil 放在管道的最后。 操作符只能看到它前面的内容，因此，放在 switchMap 上方的 takeUntil 会结束外部源，却让内部 Observable 继续运行——这是看起来像正确清理代码的内存泄漏。

将 catchError 放在内部管道中，而不是外部管道中。 错误会向上传递给订阅者，并永久终止整个链；在内部 Observable 中恢复，才能让外部流继续处理下一个值。

根据所需的并发方式选择扁平化操作符——取消上一个（switchMap）、并行运行（mergeMap）、排队（concatMap）、忙碌时忽略（exhaustMap）。在搜索框中默认使用 mergeMap，会导致过时的响应覆盖较新的响应。

</critical_requirements>

---

自动检测： RxJS、rxjs、Observable、Subject、BehaviorSubject、ReplaySubject、AsyncSubject、 subscribe、pipe、switchMap、mergeMap、concatMap、exhaustMap、combineLatest、forkJoin、 withLatestFrom、fromEvent、defer、interval、timer、catchError、retry、debounceTime、throttleTime、 distinctUntilChanged、takeUntil、shareReplay、firstValueFrom、lastValueFrom、Subscription

适用场景：

• 需要组合的事件流——防抖、节流、缓冲、去重
• 协调多个异步源——合并、并行等待、竞速、排序执行
• 可取消的工作，即新输入应放弃正在进行的请求
• 随时间到达的实时数据——套接字、服务器发送事件、轮询
• 将重试和退避策略作为管道的一部分来表达

其他部分处理：

• 没有流语义的单次请求——一行 async/await 就能表达，封装成 Observable 毫无益处
• 本地组件状态——对于只需读取和写入的值，流并不合适
• 应用状态的归属——RxJS 在不同位置之间传递值；决定状态存放在哪里是另一个问题
• 框架绑定——将流转换为视图层渲染的内容，属于该层自己的 API

---

一切都是流：点击、响应、计时器、套接字帧。Observable 描述的是工作，而不是工作本身，因此 pipe() 链会构建管道，subscribe() 才会启动它——这就是为什么没有订阅者的 pipe 什么也不做，也解释了为什么同一个 Observable 通常会在每次订阅时都执行一次工作。

这种惰性也是最常见的意外来源：在链中放入任何急切执行的内容（例如在构造时调用 fetch，而不是在 defer 中调用），都会脱离这种模型，只运行一次并被共享，无论是否有人订阅。

---

核心模式

模式 1：创建，以及保持惰性

根据源选择创建函数，并将任何急切执行的内容包装在 defer 中，这样每个订阅者都会得到独立的执行。

import {
  of,
  from,
  fromEvent,
  interval,
  timer,
  defer,
  EMPTY,
  throwError,
} from "rxjs";

of(1, 2, 3); // fixed values, then completes
from([1, 2, 3]); // any iterable, then completes
fromEvent(element, "click"); // never completes
interval(1000); // never completes
timer(2000); // one value, then completes

defer(() => from(fetch("/api/users"))); // fresh request per subscriber
from(fetch("/api/users")); // ❌ fires at construction, shared by all

完整代码：examples/core.md

---

模式 2：在管道中使用操作符，绝不嵌套订阅

const adults$ = users$.pipe(
  filter((user) => user.age >= 18),
  map((user) => user.name),
  distinctUntilChanged(),
  tap((name) => console.log(name)), // side effects live in tap
);

在一个 .subscribe() 中再调用一个 .subscribe()，会创建一个没有任何对象持有引用的订阅，因此也无法结束它——扁平化操作符正是为替代这种写法而存在的。

完整代码：examples/core.md

---

模式 3：使用 catchError 恢复错误

const results$ = query$.pipe(
  switchMap((query) =>
    defer(() => from(search(query))).pipe(
      retry({ count: 3, delay: (_, n) => timer(1000 * 2 ** (n - 1)) }),
      catchError(() => of({ results: [], failed: true })), // inner — outer survives
    ),
  ),
);

catchError 必须返回一个 Observable，而它的位置决定了哪些部分会继续运行：放在内部管道中，外部流会继续；放在外部管道中，整个链会在首次失败时终止。

完整代码：examples/core.md

---

模式 4：速率限制

// Wait for a pause — user input
input$.pipe(debounceTime(300), distinctUntilChanged());

// Cap the rate — continuous events
fromEvent(window, "scroll").pipe(throttleTime(200));

debounceTime 会在静默一段时间后发出最后一个值；throttleTime 会发出每个时间窗口中的第一个值，并丢弃其余值。在防抖之后使用 distinctUntilChanged，可以避免输入后又撤销时再次发出相同请求。

完整代码：examples/higher-order.md，位于搜索管道中。 操作符对比：reference.md

---

模式 5：扁平化操作符

当新的外层值到达时，如果内部 Observable 仍在运行，应该如何处理？这就是需要作出的选择。

Cancel the previous inner and switch    → switchMap    (search, navigation)
Run the new one alongside               → mergeMap     (uploads, fire-and-forget)
Queue it until the previous finishes    → concatMap    (ordered saves, message send)
Ignore it until the current one ends    → exhaustMap   (submit, login, refresh)

query$.pipe(switchMap((q) => search(q)));
files$.pipe(mergeMap((file) => upload(file), 3)); // 3 = concurrency limit
saves$.pipe(concatMap((data) => save(data)));
clicks$.pipe(exhaustMap(() => submit()));

完整代码：examples/higher-order.md

---

模式 6：Subject

Subject 既是 Observable，也是 Observer，因此它既可用于多播，也可用于从命令式代码中推送值。

new Subject<T>(); // no initial value, no replay
new BehaviorSubject<T>(initial); // always has a current value; .getValue()
new ReplaySubject<T>(n, windowMs); // replays the last n (optionally, recent ones)
new AsyncSubject<T>(); // the final value only, and only on complete()

BehaviorSubject 用于始终有值的状态，ReplaySubject 用于晚加入的订阅者需要的历史记录，Subject 用于没有历史记录的事件，AsyncSubject 用于单个最终结果。

完整代码：examples/subjects.md

---

模式 7：使用 takeUntil 清理

class Controller {
  private readonly destroy$ = new Subject<void>();

  start(): void {
    source$
      .pipe(
        switchMap((id) => this.load(id)),
        takeUntil(this.destroy$), // last — ends the inner subscription too
      )
      .subscribe((data) => this.render(data));
  }

  stop(): void {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

一个 destroy$ 就能结束所有经过它的订阅。完整代码： examples/memory-leaks.md

---

<red_flags>

注意事项

运行时错误：

• 未处理的错误会永久终止链。之后发出的所有值都会被静默丢弃，且无法恢复流，只能重新订阅。
• catchError 返回普通值而不是 Observable 会导致类型错误；解决方法是使用 of(fallback)。
• 将 takeUntil 放在高阶操作符之前，会结束外层源，却让内部订阅继续运行。清理代码看起来正确，但泄漏不会被察觉。
• 嵌套的 .subscribe() 调用会产生没有引用的订阅，并随着每次外层发射而不断增加。
• 如果 Observable 完成时没有发出值，firstValueFrom 会拒绝；已弃用的 .toPromise() 则会解析为 undefined，因此迁移会改变失败方式。

令人意外的行为：

• from(promise) 会在构造时运行 Promise，而不是在订阅时运行；每个订阅者都会得到相同的已完成结果。defer(() => from(promise)) 可以恢复惰性求值。
• shareReplay(1) 默认将 refCount 设为 false，因此最后一个订阅者离开后，源仍会保持订阅状态。几乎总是应该使用 shareReplay({ bufferSize: 1, refCount: true })。
• combineLatest 要等到每个源至少都发出一次值后才会发射；任何一个不发射的源都会卡住整个组合。
• forkJoin 只有在每个源都 完成 后才会发射，因此列表中有 Subject 或 interval 就意味着它永远不会发射。
• retry 会从头开始重新订阅源，因此非幂等源会再次执行其副作用。
• interval(0) 仍然是异步的——它会安排任务，而不会同步发射。
• BehaviorSubject.getValue() 无需订阅即可读取当前值，这会悄然将响应式代码变成轮询。
• 本该使用 switchMap 的地方使用了 mergeMap，会产生竞争：响应按乱序到达，且最慢的响应会胜出。

</red_flags>