乐闻世界logo
5月27日 21:26

RxJS 中如何创建自定义操作符？

核心答案

RxJS 创建自定义操作符有三种方式，优先使用组合现有操作符，其次用 pipeable 函数，避免直接 new Observable：

typescript
// 1. 组合现有操作符（推荐）
function debounceSearch(ms = 300) {
  return pipe(
    debounceTime(ms),
    distinctUntilChanged()
  );
}

// 2. pipeable 函数
function debug<T>(label: string): OperatorFunction<T, T> {
  return source$ => source$.pipe(
    tap(v => console.log(label, v))
  );
}

三种方式的区别

方式适用场景复杂度
组合现有操作符逻辑由已有操作符拼装即可低
pipeable 函数（返回 source$ => new Observable）需要精细控制订阅和退订中
直接 new Observable极端定制场景，需手动管理全部生命周期高

直接 new Observable 需要自己处理 next/error/complete 和 teardown，容易遗漏导致内存泄漏，非必要不使用。

追问：操作符内如何保证资源释放？

返回的 Observable 订阅时必须返回 teardown 函数：

typescript
function withTimeout<T>(ms: number): OperatorFunction<T, T> {
  return source$ => new Observable(subscriber => {
    const timer = setTimeout(() => subscriber.error(new Error('timeout')), ms);
    const sub = source$.subscribe({
      next: v => subscriber.next(v),
      error: e => subscriber.error(e),
      complete: () => subscriber.complete()
    });
    // teardown：取消定时器 + 退订上游
    return () => { clearTimeout(timer); sub.unsubscribe(); };
  });
}

忘记返回 teardown 是自定义操作符最常见 bug，会导致定时器、子订阅等资源无法回收。

追问：RxJS 7+ 写法有什么变化？

retryWhen、tap(next, error, complete) 三参数重载等已在 RxJS 7 中弃用或移除。自定义操作符应使用 retry({ count, delay }) 等新 API，组合操作符时优先用 pipe() 而非链式调用。

标签：Rxjs