RxJS subscription 返回重复数据且次数不断增加

结论

问题通常不在 Node.js 的 Promise 或数据库查询，而是 Angular 组件中的 subscription 不断累积。

每当 activatedRoute.params 发出参数时，回调都会重新执行，并创建一个新的 getUserApprovalsUpdateListener().subscribe(...)。旧 subscription 没有取消，所以每次调用 approvalsUpdated.next(...)，所有历史订阅者都会收到同一批数据。页面看起来就像出现了重复数据，而且重复次数会越来越多。

解决办法是在创建新订阅前取消旧订阅，或者用 switchMap 组织 Observable 流，并在组件销毁时统一清理。

为什么会出现重复

当前代码的执行过程相当于：

route params 第一次触发
└── 创建 approvals subscription A

route params 第二次触发
├── subscription A 仍然存在
└── 创建 approvals subscription B

route params 第三次触发
├── subscription A 仍然存在
├── subscription B 仍然存在
└── 创建 approvals subscription C

之后 service 执行：

this.approvalsUpdated.next(approvalsPending);

A、B、C 都会收到同一份结果。如果每个订阅回调都将结果追加到页面数组中，最终就会显示三份相同的数据。

这也解释了为什么 Node endpoint 返回的数量正确，组件中却出现重复。服务端可能只返回了一份正确结果，但同一个事件在客户端被多个订阅者处理了多次。

还要注意，service 中的这段代码不会清除已经发出的数据：

approvalsPending = [];

next() 会同步通知订阅者。重新给局部变量赋值既不会取消订阅，也不会影响订阅者已经收到的数组。

推荐方案：让 service 直接返回 HTTP Observable

这个请求本质上是一次 HTTP 查询，通常不需要再通过 Subject 转发。可以让 service 直接返回 Observable，然后在组件中使用 switchMap，根据路由参数发起请求。

Service

import { Observable } from 'rxjs';
import { map } from 'rxjs/operators';

interface ApprovalsResponse {
  message: string;
  approvals: any[];
}

getUserPendingApprovals(userid: string): Observable<any[]> {
  return this.http
    .get<ApprovalsResponse>(`/api/approvals/${userid}`)
    .pipe(
      map(response => response.approvals)
    );
}

Component

import { Subject } from 'rxjs';
import { distinctUntilChanged, map, switchMap, takeUntil } from 'rxjs/operators';

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

ngOnInit(): void {
  this.activatedRoute.params
    .pipe(
      map(params => params['userid']),
      distinctUntilChanged(),
      switchMap(userid => {
        this.userid = userid;
        return this.approvalsService.getUserPendingApprovals(userid);
      }),
      takeUntil(this.destroy$)
    )
    .subscribe(approvalsPending => {
      // 直接替换，避免在旧数组上持续追加
      this.approvalsPending = approvalsPending;
    });
}

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

switchMap 在这里有两个作用：

• 路由参数变化时，取消上一次尚未完成的 HTTP Observable。
• 每次参数变化时都不会留下新的内部 subscription。

takeUntil(this.destroy$) 可以保证组件销毁后，路由参数订阅也会被清理。

如果暂时保留 Subject 结构

如果现有项目必须继续使用 approvalsUpdated，至少要避免在路由参数回调中不断累积监听器。

一种简单的处理方式是在创建新监听器之前取消旧监听器：

ngOnInit(): void {
  this.routeParamsSub = this.activatedRoute.params.subscribe(
    (params: Params) => {
      this.userid = params['userid'];

      this.approvalsSub?.unsubscribe();

      this.approvalsSub = this.approvalsService
        .getUserPendingApprovalsUpdateListener()
        .subscribe((approvalsPending: any[]) => {
          this.approvalsPending = approvalsPending;
        });

      this.approvalsService.GetUserPendingApprovals(this.userid);
    }
  );
}

ngOnDestroy(): void {
  this.routeParamsSub?.unsubscribe();
  this.approvalsSub?.unsubscribe();
  this.currentWeekInvoicesSub?.unsubscribe();
  this.currentDayInvoicesSub?.unsubscribe();
}

最好先建立订阅监听器，再调用请求方法。虽然 Angular HttpClient 请求是异步的，原代码通常不会因此漏掉结果，但先订阅再触发可以避免依赖异步时序。

也可以将监听器移到路由订阅外部，让它在组件的整个生命周期中只创建一次：

ngOnInit(): void {
  this.approvalsSub = this.approvalsService
    .getUserPendingApprovalsUpdateListener()
    .subscribe((approvalsPending: any[]) => {
      this.approvalsPending = approvalsPending;
    });

  this.routeParamsSub = this.activatedRoute.params.subscribe(
    (params: Params) => {
      this.userid = params['userid'];
      this.approvalsService.GetUserPendingApprovals(this.userid);
    }
  );
}

这种写法适用于监听器不需要随路由参数重新创建的情况。

使用较新 Angular 的自动清理方式

如果当前 Angular 版本支持 takeUntilDestroyed，就不需要手动维护 destroy$：

import { DestroyRef, inject } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

private readonly destroyRef = inject(DestroyRef);

ngOnInit(): void {
  this.activatedRoute.params
    .pipe(
      map(params => params['userid']),
      distinctUntilChanged(),
      switchMap(userid =>
        this.approvalsService.getUserPendingApprovals(userid)
      ),
      takeUntilDestroyed(this.destroyRef)
    )
    .subscribe(approvalsPending => {
      this.approvalsPending = approvalsPending;
    });
}

是否能使用这种方式，取决于项目的 Angular 版本。

还需要检查的地方

即使订阅问题已经修复，下面的写法仍然可能导致页面数据重复：

this.approvalsPending.push(...approvalsPending);

或者：

this.approvalsPending = [
  ...this.approvalsPending,
  ...approvalsPending
];

如果每次响应返回的都是完整列表，应该直接替换：

this.approvalsPending = approvalsPending;

还可以分别记录 HTTP 请求次数和订阅回调次数：

console.count('HTTP response received');

console.count('approvals listener executed');

如果一次 HTTP 响应对应多次 listener 执行，基本可以确定是订阅累积。如果 HTTP 响应本身也出现多次，还需要检查 GetUserPendingApprovals() 是否被重复调用，以及 activatedRoute.params 是否发出了多次值。

解决这个问题的重点不是清空 approvalsPending，而是避免嵌套订阅不断创建新的长期监听器，并确保所有组件级 subscription 在不再需要时都被取消。

备注：内容仅供参考。