combineLatest未等待所有Observable发射新值的问题求助

我有两个Service，各自包含一个BehaviorSubject转成的Observable。组件需要获取两个Observable的最新发射值，但使用combineLatest时，单个Observable发生变化就会触发订阅回调，而非等待两者都完成更新后再触发。调用createUserAndSession函数后，combineLatest会在2500ms和5000ms分别输出结果，不符合预期。

代码示例 #

组件代码

export class CreateSessionComponent {
  constructor(
    private sessionService: SessionService,
    private userService: UserService
  ) {
    combineLatest([this.userService.user$, this.sessionService.session$])
      .subscribe({
        next: (data) => console.log(data),
      });
  }

  public createUserAndSession(): void {
    this.sessionService.createSession();
    this.userService.createUser();
  }
}

UserService代码

export class UserService {
  private userSubject = new BehaviorSubject<any | null>(null);
  public user$ = this.userSubject.asObservable();

  public createUser(): void {
    setTimeout(() => {
      this.userSubject.next(`User ${Math.random()} `);
    }, 5000);
  }
}

SessionService代码

export class SessionService {
  private sessionSubject = new BehaviorSubject<any | null>(null);
  public session$ = this.sessionSubject.asObservable();

  public createSession(): void {
    setTimeout(() => {
      this.sessionSubject.next(`Session ${Math.random()} `);
    }, 2500);
  }
}

原因分析 #

combineLatest的特性是：当任意一个源Observable发射新值时，它会收集所有源Observable的最新值并组合发射。这里因为使用的是BehaviorSubject，它会在订阅时立即发射当前保存的值（初始为null）。所以：

1. 2500ms时session$发射新值，此时user$的最新值还是初始的null，combineLatest会组合这两个值触发回调；
2. 5000ms时user$发射新值，此时session$的最新值是刚发射的会话值，combineLatest再次组合触发回调。

解决方案 #

方案1：使用forkJoin（适合一次性请求场景）

如果createUser和createSession是一次性操作，不需要持续订阅流的变化，可以修改Service返回Observable，用forkJoin等待两个操作都完成后再获取结果：

修改Service：

// UserService
public createUser(): Observable<string> {
  return timer(5000).pipe(map(() => `User ${Math.random()} `));
}

// SessionService
public createSession(): Observable<string> {
  return timer(2500).pipe(map(() => `Session ${Math.random()} `));
}

修改组件：

public createUserAndSession(): void {
  forkJoin([this.userService.createUser(), this.sessionService.createSession()])
    .subscribe(data => console.log(data));
}

forkJoin会等待所有源Observable完成后，发射最后一次值的组合，只会触发一次回调。

方案2：保留BehaviorSubject，过滤无效值并只取一次

如果需要保留BehaviorSubject供其他组件订阅，可在combineLatest后添加过滤和取一次的操作符：

constructor(
  private sessionService: SessionService,
  private userService: UserService
) {
  combineLatest([this.userService.user$, this.sessionService.session$])
    .pipe(
      // 过滤掉任意一个值为null的情况
      filter(([user, session]) => user !== null && session !== null),
      // 只取第一次符合条件的组合值
      take(1)
    )
    .subscribe(data => console.log(data));
}

这样只有当两个Observable都发射了非null的新值时，才会触发一次回调，符合“等待两者都完成更新”的需求。

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