如何让新订阅Observer获取Subject全量历史值及每次发布的完整序列？

看起来你需要的是一个能记住所有历史值，并且每次有新值时都把完整序列推给订阅者的Subject，对吧？单纯的ReplaySubject确实满足不了这个需求——它只会在订阅时重播历史值，之后新值到来时只发送单个新值，而不是你要的完整序列。

这里有个完美贴合你需求的解决方案：

方案1：自定义FullSequenceSubject #

我们自己封装一个Subject，内部维护所有已发布值的数组，每次调用next时更新数组并把整个序列推给所有订阅者：

import { Subject } from 'rxjs';

class FullSequenceSubject<T> extends Subject<T[]> {
  private readonly valueHistory: T[] = [];

  next(value: T): void {
    // 把新值加入历史队列
    this.valueHistory.push(value);
    // 发送数组副本（防止外部修改内部状态）
    super.next([...this.valueHistory]);
  }

  // 按需重写complete和error方法（保持默认行为即可）
  complete(): void {
    super.complete();
  }

  error(err: any): void {
    super.error(err);
  }
}

使用示例

// 初始化我们的自定义Subject
const sub = new FullSequenceSubject<number>();

sub.next(1);
sub.next(2);
sub.next(3);

// 在#hack 1处订阅
sub.subscribe(sequence => {
  console.log('订阅1收到:', sequence);
  // 输出顺序：[1,2,3] → [1,2,3,4] → [1,2,3,4,5] → ...
});

sub.next(4);
sub.next(5);
sub.next(6);

// 在#hack 2处订阅
sub.subscribe(sequence => {
  console.log('订阅2收到:', sequence);
  // 输出顺序：[1,2,3,4,5,6] → [1,2,3,4,5,6,7] → ...
});

sub.next(7);

为什么这个方案适合你？ #

• 订阅即得完整历史：不管你在#hack1还是#hack2时机订阅，都会立即拿到从第一个值到当前最新值的完整序列。
• 新值触发完整序列推送：每次发布新值，所有订阅者都会收到更新后的完整序列，而不是单个新值。
• 状态安全：发送的是历史数组的副本，避免外部代码意外修改内部维护的历史数据。

方案2：结合ReplaySubject与scan（备选） #

如果你不想自定义类，也可以用RxJS内置操作符组合实现：

import { ReplaySubject, scan, takeLast, repeat } from 'rxjs';

const source = new ReplaySubject<number>();
// 用scan累积所有值，再用takeLast(1)确保订阅时直接拿到完整序列
const fullSequence$ = source.pipe(
  scan((acc, val) => [...acc, val], [] as number[]),
  takeLast(1),
  repeat()
);

// 后续使用和你原来的逻辑一致：
source.next(1);
source.next(2);
source.next(3);

// #hack1处订阅
fullSequence$.subscribe(seq => console.log('订阅1:', seq));

source.next(4);
source.next(5);
source.next(6);

// #hack2处订阅
fullSequence$.subscribe(seq => console.log('订阅2:', seq));

source.next(7);

不过这个方案相对繁琐一点，自定义Subject的方式更直观，也更符合你原来的使用习惯。

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