RxJS中scan合并Partial数据后部分订阅场景无法获取完整对象

问题根源 #

你的问题核心是**person$默认是冷Observable**：每次新订阅（比如checkValues里用first()的订阅）都会重新订阅源personUpdates$。而personUpdates$作为BehaviorSubject，只会把当前的最新单个更新值推给新订阅者，不会重放所有历史更新记录。这就导致新订阅的scan只能拿到最后一次的Partial<Person>，和初始值合并后，无法得到之前所有更新累积的完整对象——而构造函数里的订阅是在所有更新前建立的，能接收每一次更新，所以scan可以累积所有变化。

解决方案 #

要让person$在任何场景下订阅都能拿到完整的累积对象，需要把它转为热Observable，保存最新的累积状态。推荐两种简单实现方式：

方式1：使用shareReplay(1)（最简洁）

在person$的管道中添加shareReplay(1)，它会缓存最新的一次输出结果，自动推送给所有新订阅者：

personUpdates$ = new BehaviorSubject<Partial<Person>>({});

person$ = this.personUpdates$.pipe(
  scan((acc, curr) => ({ ...acc, ...curr }), {
    name: '',
    firstName: '',
    lastName: '',
  } as Person),
  shareReplay(1) // 关键：缓存最新的累积结果
);

updatePerson = (person: Partial<Person>) => {
  this.personUpdates$.next(person);
};

方式2：手动用BehaviorSubject维护状态

直接把person$封装为BehaviorSubject，手动同步scan的累积结果：

// 私有BehaviorSubject维护内部状态
private _person$ = new BehaviorSubject<Person>({
  name: '',
  firstName: '',
  lastName: '',
});
// 对外暴露只读Observable
person$ = this._person$.asObservable();

personUpdates$ = new BehaviorSubject<Partial<Person>>({});

constructor() {
  // 订阅更新流，累积结果后同步到_person$
  this.personUpdates$.pipe(
    scan((acc, curr) => ({ ...acc, ...curr }), this._person$.value)
  ).subscribe(this._person$);
}

updatePerson = (person: Partial<Person>) => {
  this.personUpdates$.next(person);
};

两种方式都能确保，无论何时订阅person$，都能获取到所有更新合并后的完整Person对象。

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