为何使用subscribe简写初始化ReplaySubject后next不触发async管道？

我遇到一个RxJS的问题：当用subscribe(this.subject)这种简写方式把Observable的流订阅到ReplaySubject时，后续手动调用subject.next()无法触发Angular的async管道更新，只有页面初始加载时能正常工作。但用箭头函数手动调用next()的写法就完全正常。

正常工作的代码 #

export class Container implements OnInit {
  public companyDetailSubject$: ReplaySubject<CompanyDetail> = new ReplaySubject<CompanyDetail>(1);

  public ngOnInit(): void {
    this.companyControllerService.getUserCompany().subscribe((company) => this.companyDetailSubject$.next(company));
  }

  private updateCompanyName(newCompanyName: string) {
    this.companyControllerService.updateCompany({...this.companyDetail, name: newCompanyName})
      .pipe(
          tap(updatedCompany => {
            this.companyDetailSubject$.next(updatedCompany); // 这里能正常触发async管道，子组件能拿到新数据
          }),
      )
      .subscribe();
  }
}

HTML模板：

<child-component
        [companyDetail]="companyDetailSubject$ | async">
</child-component>

无法正常工作的代码（仅修改了ngOnInit里的订阅方式） #

export class Container implements OnInit {
  public companyDetailSubject$: ReplaySubject<CompanyDetail> = new ReplaySubject<CompanyDetail>(1);

  public ngOnInit(): void {
    // 唯一的区别：用简写方式订阅到ReplaySubject
    this.companyControllerService.getUserCompany().subscribe(this.companyDetailSubject$);
  }

  private updateCompanyName(newCompanyName: string) {
    this.companyControllerService.updateCompany({...this.companyDetail, name: newCompanyName})
      .pipe(
          tap(updatedCompany => {
            this.companyDetailSubject$.next(updatedCompany); // 这里调用next无法触发async管道，子组件拿不到新数据
          }),
      )
      .subscribe();
  }
}

原因解析 #

问题出在subscribe(this.subject)这种简写方式的行为上：当你把一个Subject直接传给subscribe()方法时，RxJS会自动帮你完成订阅+自动管理Subject的完结/错误状态。

具体来说，当getUserCompany()返回的Observable执行complete()时，RxJS会自动调用companyDetailSubject$.complete()。而已经完结的Subject不再会发射任何新值，所以后续你手动调用next()时，这个Subject已经处于完结状态，async管道自然不会收到新的更新。

而第一种用箭头函数的写法里，你只手动调用了next()，并没有把Observable的complete()事件传递给Subject，所以Subject始终处于活跃状态，后续的next()调用能正常触发async管道。

解决办法 #

如果你想保留简写方式，同时避免Subject被自动完结，可以用以下两种方式：

• 用tap代替subscribe，这样不会传递complete/error事件：

this.companyControllerService.getUserCompany().pipe(
  tap(this.companyDetailSubject$)
).subscribe();

• 手动指定订阅回调，不处理complete事件：

this.companyControllerService.getUserCompany().subscribe({
  next: this.companyDetailSubject$.next.bind(this.companyDetailSubject$),
  error: this.companyDetailSubject$.error.bind(this.companyDetailSubject$)
  // 不处理complete事件，避免Subject被完结
});

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