使用 shareReplay 修复 rxjs 订阅上的竞争条件

Question

我正在尝试订阅一个可观察对象并记录它的值。

流动：

this.myObservable$ = new Subject<string>();
this.myObservable$.next('hello world') //in (angular) service A 

等几秒钟

let newObservable$ = this.myObservable$.asObservable().subscribe(message => console.log(message)); //subscribe in service B, nothing happens

但这不起作用，因为我订阅得太晚了。但即使使用 shareReplay 这也不起作用。

let newObservable$ = this.myObservable$.asObservable()
      .pipe(
        shareReplay(1),
      )
      .subscribe(message => console.log(message)); //also nothing happens

谁能帮助我理解为什么我不能“shareReplay”这个可观察的并获得最后发出的值？

Answer 1

原因shareReplay对您不起作用是因为您将其应用于派生的可观察量，而不是源。

您应该将其应用到源：

this.mySubject$ = new Subject<string>();
this.myObservable$ = this.mySubject.asObservable.pipe(shareReplay(1));

现在，您的新可观察值将获得重播值：

let newObservable$ = this.myObservable$.subscribe(
   message => console.log(message)
);

shareReplay就您而言，您在 的定义中应用了newObservable，这将无法实现您想要的效果。

正如其他人提到的，由于您已经有一个主题作为来源，您可以简单地使用ReplaySubject：

this.mySubject$ = new ReplaySubject<string>(1);

通常，在 Angular 服务的上下文中，您将保持Subject服务的私有性，并向消费者公开公开只读可观察值。像这样的东西：

服务：

private subject$ = new Subject();
public readonly value$ = this.subject$.pipe(shareReplay()) // if you use .pipe(), you don't really need .asObservable()

成分：

data$ = this.service.value$;

由于shareReplay是在 source ( value$) observable 上定义的，因此组件将始终获得重放的发射。