CombineLatest未取消订阅引发重复调用问题，求解决方案

你的问题根源在于每次调用createUserAndSession()时都会创建一个新的combineLatest订阅，但之前的订阅没有被取消，第二次调用时旧订阅仍会监听user$和session$的变化，导致重复执行connectHostToSession。下面是几种可行的解决方式：

方法一：手动管理订阅 #

在组件类中声明一个Subscription对象，每次调用函数前先取消旧订阅，再创建新订阅：

import { Subscription } from 'rxjs';

// 组件类中添加属性
private userSessionSubscription?: Subscription;

public createUserAndSession(): void {
  // 先取消之前的订阅
  this.userSessionSubscription?.unsubscribe();

  this.userService.createUser(this.form);
  this.sessionService.createSession(this.form);

  // 保存新订阅
  this.userSessionSubscription = combineLatest([
    this.userService.user$,
    this.sessionService.session$
  ]).subscribe({
    next: ([user, session]) => {
      console.log(user, session);
      this.sessionService.connectHostToSession(user!, session);
    }
  });
}

方法二：使用take(1)自动完成订阅 #

如果你的场景只需要在user$和session$各发射一次值后就停止监听，可以用take(1)操作符，它会在收到第一个值后自动取消订阅：

import { combineLatest, take } from 'rxjs';

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

  combineLatest([
    this.userService.user$,
    this.sessionService.session$
  ]).pipe(
    take(1) // 只取一次值后自动完成订阅
  ).subscribe({
    next: ([user, session]) => {
      console.log(user, session);
      this.sessionService.connectHostToSession(user!, session);
    }
  });
}

方法三：优化逻辑，关联创建操作与结果监听 #

更严谨的做法是直接监听createUser和createSession这两个异步操作的结果，而不是依赖全局的user$和session$，避免全局流的干扰：

import { forkJoin } from 'rxjs';

public createUserAndSession(): void {
  // 同时发起创建用户和会话的请求，等待两者完成
  forkJoin([
    this.userService.createUser(this.form),
    this.sessionService.createSession(this.form)
  ]).subscribe({
    next: ([user, session]) => {
      console.log(user, session);
      this.sessionService.connectHostToSession(user!, session);
    }
  });
}

注：如果createUser和createSession是HTTP请求这类单次发射的Observable，forkJoin会在两者都完成后返回结果，比combineLatest更适合这种场景。

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