RxJS Observable结合async pipe无法订阅，页面始终显示#noMatch

核心问题 #

• reduce操作符的误用：reduce会等待源Observable完全完成后才发射最终累加结果。而authService.user$通常是持续发射值的Observable（比如用户登录状态变更时会持续推送），永远不会触发完成，导致filteredMatches$从未发射任何值，async pipe拿不到数据，因此始终显示noMatch模板。
• 数据类型错误：matches$发射的是User[]类型，但原代码里直接访问curr.type——curr是数组，不是单个User对象，这会导致逻辑错误。
• 潜在竞态问题：通过组件属性this.type传递用户类型，当user$频繁发射新值时，可能出现后续管道使用过时type的情况。

修正后的TS代码 #

user$: Observable<User> = this.authService.user$;
matches$: Observable<{ type: string; users: User[] }> = this.user$.pipe(
  switchMap((user: User) => {
    // 将type和用户数组一起传递，避免竞态
    return of({ type: user.type, users: [user, user, user] });
  }),
  // 自定义比较逻辑，避免不必要的发射
  distinctUntilChanged((prev, curr) => 
    prev.type === curr.type && JSON.stringify(prev.users) === JSON.stringify(curr.users)
  ),
);

filteredMatches$: Observable<{ type1: User[]; type2: User[] }> = this.matches$.pipe(
  // 用scan替代reduce，每次源发射值就更新并返回累加结果
  scan((acc, { type, users }) => {
    console.log('in the pipe');
    // 遍历每个用户进行分类
    users.forEach(user => {
      this.networkingService.decideType(type, user.type)
        ? acc.type1.push(user)
        : acc.type2.push(user);
    });
    return acc;
  }, { type1: [], type2: [] }), // 初始化累加器
);

ngOnInit(): void {
  // 无需再重新赋值user$，直接在定义时绑定authService.user$即可
}

修正后的HTML代码 #

<ng-template *ngIf="(filteredMatches$ | async) as matches; else noMatch">
  <app-child *ngFor="let user of matches.type1" [user]="user"></app-child>
  <app-child *ngFor="let user of matches.type2" [user]="user"></app-child>
</ng-template>

<ng-template #noMatch>
  <app-no-match></app-no-match>
</ng-template>

关键修改说明 #

• 用scan替代reduce：scan会在源Observable每次发射值时就触发累加逻辑，并立即返回当前的累加结果，完美适配持续发射的数据流场景。
• 传递完整上下文：在switchMap里返回包含type和users的对象，避免依赖组件属性传递状态，消除竞态风险。
• 修复数组遍历逻辑：遍历users数组中的每个User对象进行类型判断，而不是错误地访问数组的type属性。
• 优化distinctUntilChanged：自定义比较逻辑，确保只有当type或用户数组真正变化时才发射新值，减少不必要的管道执行。

内容的提问来源于stack exchange，提问作者Igor Cantele