RxJS 使用时的取消订阅问题

参考资料： https://www.thisdot.co/blog/best-practices-for-managing-rxjs-subscriptions

在刚开始学习使用 RxJS 时，如果没有仔细考虑取消订阅，那么很容易写出 bug ，请思考以下问题：

在每次写下 xxx$.subscribe() 后，我们需要取消订阅吗？
哪些场景可以不取消订阅？
哪些场景不取消订阅会有 bug ？

没有取消订阅会有什么问题

举个例子，在进入页面 A 后，有个数据需要每 3 秒就请求一次，代码如下：

@Component({
  selector: 'app-playground',
  standalone: true,
  imports: [CommonModule],
  template: `<div>{{ date }}</div>`,
})
export class PlaygroundComponent {
  date = '';

  constructor(private http: HttpClient) {}

  ngOnInit(): void {
    this.http
      .get('http://localhost:3000/date', { responseType: 'text' })
      // 通过 repeat 来实现 3s 的轮询
      .pipe(repeat({ delay: 3000 }))
      .subscribe((res) => {
        this.date = res;
      });
  }
}

上面的代码中我们只写了 .subscribe 没有考虑取消订阅。当我们访问了页面 A 然后又通过前端路由跳转其它页面后，这里的轮询请求并不会停止，因为我们并没有取消订阅。

那么哪些场景需要处理取消订阅呢？

当这个 observable 会持续产生值时。如果不取消订阅，假设我们是在组件挂载时订阅，那么通过路由跳转可以使组件不停卸载和挂载，那么会有越来越多的 observable ，也许对页面功能并不会有啥影响，但是程序的内存占用会慢慢变大？
当 .subscribe 里面有“副作用”时。比如我们在页面 A 请求了一个数据，拿到数据后会跳转特定路由，如果离开页面 A 时没有取消订阅，那么假设用户访问了页面 A 然后在请求完成前就离开了页面 A ，但是由于没有取消订阅，那么 .subscribe 里面的代码还是会执行，用户会被跳转到其它页面去。

当然，实际开发中，我们也可以为了省事，无论何时写 .subscribe 都加上取消订阅的处理。

能怎么减少需要取消订阅的场景吗

在 Angular 中，我们可以尽量在 html 中通过 async pipe 来直接使用 observable ，这样 html 销毁时 Angular 会处理取消订阅，我们就不用手动处理了，如下的代码就不需要再手动取消订阅了。

When the component gets destroyed, the async pipe unsubscribes automatically to avoid potential memory leaks.

@Component({
  selector: 'app-playground',
  standalone: true,
  imports: [CommonModule],
  // 使用 async pipe 时， Angular 会为我们处理取消订阅
  template: `<div>{{ date$ | async }}</div>`,
})
export class PlaygroundComponent {
  date$ = this.http
    .get('http://localhost:3000/date', { responseType: 'text' })
    .pipe(repeat({ delay: 1000 }));

  constructor(private http: HttpClient) {}
}

确实需要手动取消订阅时，怎么写

unsubscribe 怎么写比较优雅呢，我们可以参考 antd-angular 中写法，见 ng-zorro-antd/empty.component.ts 。

// 1. 定义一个 destroy$
private readonly destroy$ = new Subject<void>();

// 2. 当需要 .subscribe() 时，先加上 .pipe(takeUntil(this.destroy$))
this.xxx$.pipe(takeUntil(this.destroy$)).subscribe(() => {
  // xxx
});

// 3. ngOnDestroy 中如下写
ngOnDestroy(): void {
  this.destroy$.next();
  this.destroy$.complete();
}

你会不会想，如果我 subscribe 的是一个很复杂的 observable ，比如 merge(observable1, observable2).subscribe() 这种涉及了多个 observable 的，使用 takeUntil 会把相关的 observable 都取消订阅吗？请放心！它会的！下面是个测试例子，当涉及多个 observable 时， createInterval 里面的 unsubscribe 都会执行到。

const createInterval = (time: number) => {
  let count = 0
  return new Observable(function subscribe(subscriber) {
    const intervalId = setInterval(() => {
      count = count + 1
      console.log('interval: ', time, count)
      subscriber.next(count)
    }, time)
    console.log('subscribe: ', time)

    return function unsubscribe() {
      clearInterval(intervalId)
      console.log('unsubscribe: ', time)
    }
  })
}

const destroy$ = new Subject<void>()

// first example
merge(createInterval(1000), createInterval(400))
  .pipe(takeUntil(destroy$))
  .subscribe()

// second example
createInterval(1000)
  .pipe(combineLatestWith(createInterval(400)), takeUntil(destroy$))
  .subscribe()

setTimeout(() => {
  destroy$.next()
  destroy$.complete()
}, 2000)

总结

写下每个 .subscribe() 时要想清楚需要做取消订阅的处理不
在 Angular 中，尽量直接在 html 中使用 date$ | async ，这样就不用手动取消订阅了
确实需要取消订阅时用 takeUntil(destroy$) 即可

findxc / blog